payments/internal/connectors/engine/engine.go

package engine

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strings"
	"sync"
	"time"

	"github.com/hanzo-fi/go-libs/v5/pkg/observe/log"
	"github.com/hanzo-fi/go-libs/v5/pkg/query"
	"github.com/hanzo-fi/go-libs/v5/pkg/storage/bun/paginate"
	"github.com/hanzo-fi/go-libs/v5/pkg/types/pointer"
	"github.com/hanzo-fi/payments/internal/connectors"
	"github.com/hanzo-fi/payments/internal/connectors/engine/utils"
	"github.com/hanzo-fi/payments/internal/connectors/engine/workflow"
	"github.com/hanzo-fi/payments/internal/connectors/plugins/registry"
	"github.com/hanzo-fi/payments/internal/otel"
	"github.com/hanzo-fi/payments/internal/storage"
	errorsutils "github.com/hanzo-fi/payments/pkg/domain/errors"
	"github.com/hanzo-fi/payments/pkg/domain/models"
	"github.com/go-playground/validator/v10"
	"github.com/google/uuid"
	"github.com/pkg/errors"
	"go.temporal.io/api/enums/v1"
	"go.temporal.io/sdk/client"
	"golang.org/x/sync/errgroup"
)

//go:generate mockgen -source engine.go -destination engine_generated.go -package engine . Engine
type Engine interface {
	// Install a connector with the given provider and configuration.
	InstallConnector(ctx context.Context, provider string, rawConfig json.RawMessage) (models.ConnectorID, error)
	// Uninstall a connector with the given ID.
	UninstallConnector(ctx context.Context, connectorID models.ConnectorID) (models.Task, error)
	// Reset a connector with the given ID, by uninstalling and reinstalling it.
	ResetConnector(ctx context.Context, connectorID models.ConnectorID) (models.Task, error)
	// Update a connector with the given configuration.
	UpdateConnector(ctx context.Context, connectorID models.ConnectorID, rawConfig json.RawMessage) error

	// Create a Formance account, no call to the plugin, just a creation
	// of an account in the database related to the provided connector id.
	CreateFormanceAccount(ctx context.Context, account models.Account) error
	// Create a Formance payment, no call to the plugin, just a creation
	// of a payment in the database related to the provided connector id.
	CreateFormancePayment(ctx context.Context, payment models.Payment) error
	// Create a Formance payment initiation, no call to the plugin, just a creation
	// of a payment initiation in the database and the sending of the related event.
	CreateFormancePaymentInitiation(ctx context.Context, paymentInitiation models.PaymentInitiation, adj models.PaymentInitiationAdjustment) error

	// Forward a bank account to the given connector, which will create it
	// in the external system (PSP).
	ForwardBankAccount(ctx context.Context, ba models.BankAccount, connectorID models.ConnectorID, waitResult bool) (models.Task, error)
	// Create a transfer between two accounts on the given connector (PSP).
	CreateTransfer(ctx context.Context, piID models.PaymentInitiationID, attempt int, waitResult bool) (models.Task, error)
	// Reverse a transfer on the given connector (PSP).
	ReverseTransfer(ctx context.Context, reversal models.PaymentInitiationReversal, waitResult bool) (models.Task, error)
	// Create a payout on the given connector (PSP).
	CreatePayout(ctx context.Context, piID models.PaymentInitiationID, attempt int, waitResult bool) (models.Task, error)
	// Reverse a payout on the given connector (PSP).
	ReversePayout(ctx context.Context, reversal models.PaymentInitiationReversal, waitResult bool) (models.Task, error)

	// Create a user on the given connector (PSP).
	ForwardPaymentServiceUser(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID) error
	// Delete a payment service user
	DeletePaymentServiceUser(ctx context.Context, psuID uuid.UUID) (models.Task, error)
	// Delete a payment service user on a given connector (PSP)
	DeletePaymentServiceUserConnector(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID) (models.Task, error)
	// Delete a payment service user specific connection on the given connector (PSP)
	DeletePaymentServiceUserConnection(ctx context.Context, connectorID models.ConnectorID, psuID uuid.UUID, connectionID string) (models.Task, error)
	// Create a payment service user link on the given connector (PSP).
	CreatePaymentServiceUserLink(ctx context.Context, applicationName string, psuID uuid.UUID, connectorID models.ConnectorID, idempotencyKey *uuid.UUID, ClientRedirectURL *string) (string, string, error)
	// Create a payment service user update link on the given connector (PSP).
	UpdatePaymentServiceUserLink(ctx context.Context, applicationName string, psuID uuid.UUID, connectorID models.ConnectorID, connectionID string, idempotencyKey *uuid.UUID, ClientRedirectURL *string) (string, string, error)
	// Complete a payment service user link on the given connector (PSP).
	CompletePaymentServiceUserLink(ctx context.Context, connectorID models.ConnectorID, attemptID uuid.UUID, httpCallInformation models.HTTPCallInformation) error

	// We received a webhook, handle it by calling the corresponding plugin to
	// translate it to a formance object and store it.
	HandleWebhook(ctx context.Context, url string, urlPath string, webhook models.Webhook) error

	// Create a Formance pool composed of accounts.
	CreatePool(ctx context.Context, pool models.Pool) error
	// Update Pool Query
	UpdatePoolQuery(ctx context.Context, id uuid.UUID, query map[string]any) error
	// Add an account to a Formance pool.
	AddAccountToPool(ctx context.Context, id uuid.UUID, accountID models.AccountID) error
	// Remove an account from a Formance pool.
	RemoveAccountFromPool(ctx context.Context, id uuid.UUID, accountID models.AccountID) error
	// Delete a Formance pool.
	DeletePool(ctx context.Context, poolID uuid.UUID) error

	// Called when the engine is starting, to start all the connectors.
	OnStart(ctx context.Context) error
	// Called when the engine is stopping, to stop all the connectors.
	OnStop(ctx context.Context)
}

type engine struct {
	logger logging.Logger

	temporalClient client.Client
	storage        storage.Storage

	// connectors is only really present in engine to allow validation of plugin configs prior to insert into the DB
	// other plugin-side work should be performed inside workers and not directly in the engine since we don't
	// have a listener function that checks for plugin updates or installs
	connectors connectors.Manager

	stack          string
	stackPublicURL string

	wg sync.WaitGroup
}

func New(
	logger logging.Logger,
	temporalClient client.Client,
	storage storage.Storage,
	connectors connectors.Manager,
	stack string,
	stackPublicURL string,
) Engine {
	return &engine{
		logger:         logger,
		temporalClient: temporalClient,
		storage:        storage,
		connectors:     connectors,
		stack:          stack,
		wg:             sync.WaitGroup{},
		stackPublicURL: stackPublicURL,
	}
}

func (e *engine) InstallConnector(ctx context.Context, provider string, rawConfig json.RawMessage) (models.ConnectorID, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.InstallConnector")
	defer span.End()

	connectorID := models.ConnectorID{
		Reference: uuid.New(),
		Provider:  provider,
	}

	connector := models.Connector{
		ConnectorBase: models.ConnectorBase{
			ID:       connectorID,
			Provider: provider,
		},
		Config:               rawConfig,
		ScheduledForDeletion: false,
	}
	connectorName, validatedConfig, err := e.connectors.Load(connector, false, true)
	if err != nil {
		otel.RecordError(span, err)
		if _, ok := err.(validator.ValidationErrors); ok || errors.Is(err, models.ErrInvalidConfig) {
			return models.ConnectorID{}, errorsutils.NewWrappedError(err, ErrValidation)
		}
		return models.ConnectorID{}, err
	}

	connector = models.Connector{
		ConnectorBase: models.ConnectorBase{
			ID:        connectorID,
			Name:      connectorName,
			CreatedAt: time.Now().UTC(),
			Provider:  provider,
		},
		Config: validatedConfig,
	}

	// Detached the context to avoid being in a weird state if request is
	// cancelled in the middle of the operation.
	detachedCtx := context.WithoutCancel(ctx)
	// Since we detached the context, we need to wait for the operation to finish
	// even if the app is shutting down gracefully.
	e.wg.Add(1)
	defer e.wg.Done()

	if err := e.storage.ConnectorsInstall(detachedCtx, connector, nil); err != nil {
		otel.RecordError(span, err)
		return models.ConnectorID{}, err
	}

	config, err := e.connectors.GetConfig(connector.ID)
	if err != nil {
		otel.RecordError(span, err)
		return models.ConnectorID{}, err
	}

	// Launch the workflow
	run, err := e.launchInstallWorkflow(detachedCtx, connector, config)
	if err != nil {
		otel.RecordError(span, err)
		return models.ConnectorID{}, err
	}

	// Wait for installation to complete in order to return connector ID through API
	if err := run.Get(ctx, nil); err != nil {
		otel.RecordError(span, err)
		return models.ConnectorID{}, err
	}

	return connector.ID, nil
}

func (e *engine) UninstallConnector(ctx context.Context, connectorID models.ConnectorID) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.UninstallConnector")
	defer span.End()

	detachedCtx := context.WithoutCancel(ctx)
	// Since we detached the context, we need to wait for the operation to finish
	// even if the app is shutting down gracefully.
	e.wg.Add(1)
	defer e.wg.Done()

	if err := e.storage.ConnectorsScheduleForDeletion(detachedCtx, connectorID); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	now := time.Now()
	id := e.taskIDReferenceFor(IDPrefixConnectorUninstall, connectorID, "")
	task := models.Task{
		// Do not fill the connector ID as it will be deleted
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: connectorID,
		},
		Status:    models.TASK_STATUS_PROCESSING,
		CreatedAt: now,
		UpdatedAt: now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	// Launch the uninstallation in background
	_, err := e.temporalClient.ExecuteWorkflow(
		detachedCtx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connectorID.String(),
			},
		},
		workflow.RunUninstallConnector,
		workflow.UninstallConnector{
			ConnectorID:       connectorID,
			DefaultWorkerName: GetDefaultTaskQueue(e.stack),
			TaskID:            &task.ID,
		},
	)
	if err != nil {
		task.Status = models.TASK_STATUS_FAILED
		task.UpdatedAt = time.Now()
		if err := e.storage.TasksUpsert(ctx, task); err != nil {
			e.logger.Errorf("failed to update task status to failed: %v", err)
		}

		otel.RecordError(span, err)
		return models.Task{}, err
	}

	return task, nil
}

func (e *engine) ResetConnector(ctx context.Context, connectorID models.ConnectorID) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.ResetConnector")
	defer span.End()

	detachedCtx := context.WithoutCancel(ctx)
	// Since we detached the context, we need to wait for the operation to finish
	// even if the app is shutting down gracefully.
	e.wg.Add(1)
	defer e.wg.Done()

	now := time.Now()
	id := e.taskIDReferenceFor(IDPrefixConnectorReset, connectorID, "")
	task := models.Task{
		// Do not fill the connector ID as it will be deleted
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: connectorID,
		},
		Status:    models.TASK_STATUS_PROCESSING,
		CreatedAt: now,
		UpdatedAt: now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	_, err := e.temporalClient.ExecuteWorkflow(
		detachedCtx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connectorID.String(),
			},
		},
		workflow.RunResetConnector,
		workflow.ResetConnector{
			ConnectorID:       connectorID,
			DefaultWorkerName: GetDefaultTaskQueue(e.stack),
			TaskID:            task.ID,
		},
	)
	if err != nil {
		task.Status = models.TASK_STATUS_FAILED
		task.UpdatedAt = time.Now()
		if err := e.storage.TasksUpsert(ctx, task); err != nil {
			e.logger.Errorf("failed to update task status to failed: %v", err)
		}

		otel.RecordError(span, err)
		return models.Task{}, err
	}

	return task, nil
}

func (e *engine) UpdateConnector(ctx context.Context, connectorID models.ConnectorID, rawConfig json.RawMessage) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.UpdateConnector")
	defer span.End()

	connector, err := e.storage.ConnectorsGet(ctx, connectorID)
	if err != nil {
		otel.RecordError(span, err)
		if errors.Is(err, storage.ErrNotFound) {
			return fmt.Errorf("connector %w", ErrNotFound)
		}
	}

	if connector == nil {
		err := fmt.Errorf("connector %w", ErrNotFound)
		otel.RecordError(span, err)
		return err
	}

	connector.Config = rawConfig
	connectorName, validatedConfig, err := e.connectors.Load(*connector, true, true)
	if err != nil {
		otel.RecordError(span, err)
		if _, ok := err.(validator.ValidationErrors); ok || errors.Is(err, models.ErrInvalidConfig) {
			return errorsutils.NewWrappedError(err, ErrValidation)
		}
		return err
	}
	connector.Name = connectorName
	connector.Config = validatedConfig

	if err := e.storage.ConnectorsConfigUpdate(ctx, *connector); err != nil {
		otel.RecordError(span, err)
		return err
	}

	config, err := e.connectors.GetConfig(connector.ID)
	if err != nil {
		return err
	}

	_, err = e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       fmt.Sprintf("update-schedule-polling-period-%s-%s", e.stack, connector.ID.String()),
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: true,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connector.ID.String(),
			},
		},
		workflow.RunUpdateSchedulePollingPeriod,
		workflow.UpdateSchedulePollingPeriod{
			ConnectorID: connector.ID,
			Config:      config,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}
	return nil
}

func (e *engine) CreateFormanceAccount(ctx context.Context, account models.Account) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.CreateFormanceAccount")
	defer span.End()

	provider := models.ToV3Provider(account.ConnectorID.Provider)
	capabilities, err := registry.GetCapabilities(provider)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	found := false
	for _, c := range capabilities {
		if c == models.CAPABILITY_ALLOW_FORMANCE_ACCOUNT_CREATION {
			found = true
			break
		}
	}

	if !found {
		err := &ErrConnectorCapabilityNotSupported{Capability: "CreateFormanceAccount", Provider: provider}
		otel.RecordError(span, err)
		return err
	}

	// AccountsUpsert now handles outbox events transactionally
	if err := e.storage.AccountsUpsert(ctx, []models.Account{account}); err != nil {
		otel.RecordError(span, err)
		return err
	}

	return nil
}

func (e *engine) CreateFormancePayment(ctx context.Context, payment models.Payment) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.CreateFormancePayment")
	defer span.End()

	provider := models.ToV3Provider(payment.ConnectorID.Provider)
	capabilities, err := registry.GetCapabilities(provider)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	found := false
	for _, c := range capabilities {
		if c == models.CAPABILITY_ALLOW_FORMANCE_PAYMENT_CREATION {
			found = true
			break
		}
	}

	if !found {
		err := &ErrConnectorCapabilityNotSupported{Capability: "CreateFormancePayment", Provider: provider}
		otel.RecordError(span, err)
		return err
	}

	if err := e.storage.PaymentsUpsert(ctx, []models.Payment{payment}); err != nil {
		otel.RecordError(span, err)
		return err
	}

	// Payment events are now sent via outbox pattern in PaymentsUpsert
	return nil
}

func (e *engine) CreateFormancePaymentInitiation(ctx context.Context, pi models.PaymentInitiation, adj models.PaymentInitiationAdjustment) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.CreateFormancePaymentInitiation")
	defer span.End()

	if err := e.storage.PaymentInitiationsInsert(ctx, pi, adj); err != nil {
		otel.RecordError(span, err)
		return err
	}

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

func (e *engine) ForwardBankAccount(ctx context.Context, ba models.BankAccount, connectorID models.ConnectorID, waitResult bool) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.ForwardBankAccount")
	defer span.End()

	if _, err := e.storage.ConnectorsGet(ctx, connectorID); err != nil {
		otel.RecordError(span, err)
		if errors.Is(err, storage.ErrNotFound) {
			return models.Task{}, fmt.Errorf("connector %w", ErrNotFound)
		}
		return models.Task{}, err
	}

	id := e.taskIDReferenceFor(IDPrefixBankAccountCreate, connectorID, ba.ID.String())
	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: connectorID,
		},
		ConnectorID: &connectorID,
		Status:      models.TASK_STATUS_PROCESSING,
		CreatedAt:   now,
		UpdatedAt:   now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	run, err := e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connectorID.String(),
			},
		},
		workflow.RunCreateBankAccount,
		workflow.CreateBankAccount{
			TaskID:      task.ID,
			ConnectorID: connectorID,
			BankAccount: ba,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	if waitResult {
		// Wait for bank account creation to complete
		if err := run.Get(ctx, nil); err != nil {
			otel.RecordError(span, err)
			return models.Task{}, err
		}
	}

	return task, nil
}

func (e *engine) getPayoutTaskQueue(connectorID models.ConnectorID) string {
	plugin, err := e.connectors.Get(connectorID)
	if err != nil {
		return GetDefaultTaskQueue(e.stack)
	}
	if throttle, ok := plugin.(models.PluginWithPayoutThrottle); ok && throttle.PayoutsPerSecond() > 0 {
		return GetPayoutTaskQueue(e.stack, connectorID)
	}
	return GetDefaultTaskQueue(e.stack)
}

func (e *engine) CreateTransfer(ctx context.Context, piID models.PaymentInitiationID, attempt int, waitResult bool) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.CreateTransfer")
	defer span.End()

	id := models.TaskIDReference(fmt.Sprintf("create-transfer-%s-%d", e.stack, attempt), piID.ConnectorID, piID.String())

	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: piID.ConnectorID,
		},
		ConnectorID: &piID.ConnectorID,
		Status:      models.TASK_STATUS_PROCESSING,
		CreatedAt:   now,
		UpdatedAt:   now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	run, err := e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                e.getPayoutTaskQueue(piID.ConnectorID),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: piID.ConnectorID.String(),
			},
		},
		workflow.RunCreateTransfer,
		workflow.CreateTransfer{
			TaskID:              task.ID,
			ConnectorID:         piID.ConnectorID,
			PaymentInitiationID: piID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	if waitResult {
		// Wait for bank account creation to complete
		if err := run.Get(ctx, nil); err != nil {
			otel.RecordError(span, err)
			return models.Task{}, handleWorkflowError(err)
		}
	}

	return task, nil
}

func (e *engine) ReverseTransfer(ctx context.Context, reversal models.PaymentInitiationReversal, waitResult bool) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.ReverseTransfer")
	defer span.End()

	detachedCtx := context.WithoutCancel(ctx)
	e.wg.Add(1)
	defer e.wg.Done()

	// The reversal.ID already uniquely identifies this reversal (reference +
	// connector); including reversal.CreatedAt.String() here only bloated the
	// Temporal workflow id past its length limit (EN-1346/EN-1347).
	id := models.TaskIDReference(fmt.Sprintf("reverse-transfer-%s", e.stack), reversal.ConnectorID, reversal.ID.String())
	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: reversal.ConnectorID,
		},
		ConnectorID: &reversal.ConnectorID,
		Status:      models.TASK_STATUS_PROCESSING,
		CreatedAt:   now,
		UpdatedAt:   now,
	}
	if err := e.storage.TasksUpsert(detachedCtx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	run, err := e.temporalClient.ExecuteWorkflow(
		detachedCtx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: reversal.ConnectorID.String(),
			},
		},
		workflow.RunReverseTransfer,
		workflow.ReverseTransfer{
			TaskID:                      task.ID,
			ConnectorID:                 reversal.ConnectorID,
			PaymentInitiationReversalID: reversal.ID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	if waitResult {
		// Wait for bank account creation to complete
		// use ctx instead of detachedCtx to allow the caller to cancel the operation
		// and not wait for the result
		if err := run.Get(ctx, nil); err != nil {
			otel.RecordError(span, err)
			return models.Task{}, err
		}
	}

	return task, nil
}

func (e *engine) CreatePayout(ctx context.Context, piID models.PaymentInitiationID, attempt int, waitResult bool) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.CreatePayout")
	defer span.End()

	id := models.TaskIDReference(fmt.Sprintf("create-payout-%s-%d", e.stack, attempt), piID.ConnectorID, piID.String())

	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: piID.ConnectorID,
		},
		ConnectorID: &piID.ConnectorID,
		Status:      models.TASK_STATUS_PROCESSING,
		CreatedAt:   now,
		UpdatedAt:   now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	run, err := e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                e.getPayoutTaskQueue(piID.ConnectorID),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: piID.ConnectorID.String(),
			},
		},
		workflow.RunCreatePayout,
		workflow.CreatePayout{
			TaskID:              task.ID,
			ConnectorID:         piID.ConnectorID,
			PaymentInitiationID: piID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	if waitResult {
		// Wait for bank account creation to complete
		if err := run.Get(ctx, nil); err != nil {
			otel.RecordError(span, err)
			return models.Task{}, handleWorkflowError(err)
		}
	}

	return task, nil
}

func (e *engine) ReversePayout(ctx context.Context, reversal models.PaymentInitiationReversal, waitResult bool) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.ReversePayout")
	defer span.End()

	detachedCtx := context.WithoutCancel(ctx)
	e.wg.Add(1)
	defer e.wg.Done()

	// The reversal.ID already uniquely identifies this reversal (reference +
	// connector); including reversal.CreatedAt.String() here only bloated the
	// Temporal workflow id past its length limit (EN-1346/EN-1347).
	id := models.TaskIDReference(fmt.Sprintf("reverse-payout-%s", e.stack), reversal.ConnectorID, reversal.ID.String())
	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: reversal.ConnectorID,
		},
		ConnectorID: &reversal.ConnectorID,
		Status:      models.TASK_STATUS_PROCESSING,
		CreatedAt:   now,
		UpdatedAt:   now,
	}

	if err := e.storage.TasksUpsert(detachedCtx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	run, err := e.temporalClient.ExecuteWorkflow(
		detachedCtx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: reversal.ConnectorID.String(),
			},
		},
		workflow.RunReversePayout,
		workflow.ReversePayout{
			TaskID:                      task.ID,
			ConnectorID:                 reversal.ConnectorID,
			PaymentInitiationReversalID: reversal.ID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	if waitResult {
		// Wait for bank account creation to complete
		// use ctx instead of detachedCtx to allow the caller to cancel the operation
		// and not wait for the result
		if err := run.Get(ctx, nil); err != nil {
			otel.RecordError(span, err)
			return models.Task{}, err
		}
	}

	return task, nil
}

func (e *engine) ForwardPaymentServiceUser(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.ForwardPaymentServiceUser")
	defer span.End()

	_, err := e.storage.OpenBankingForwardedUserGet(ctx, psuID, connectorID)
	switch {
	case err == nil:
		err := fmt.Errorf("user already exists on this connector: %w", ErrValidation)
		otel.RecordError(span, err)
		return err
	case err != nil && !errors.Is(err, storage.ErrNotFound):
		otel.RecordError(span, err)
		return err
	default:
	}

	psu, err := e.storage.PaymentServiceUsersGet(ctx, psuID)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	detachedCtx := context.WithoutCancel(ctx)
	e.wg.Add(1)
	defer e.wg.Done()

	plugin, err := e.connectors.Get(connectorID)
	if err != nil {
		otel.RecordError(span, err)
		if errors.Is(err, connectors.ErrNotFound) {
			return fmt.Errorf("connector %w", ErrNotFound)
		}
		return err
	}

	resp, err := plugin.CreateUser(ctx, models.CreateUserRequest{
		PaymentServiceUser: models.ToPSPPaymentServiceUser(psu),
	})
	if err != nil {
		otel.RecordError(span, err)
		return handlePluginErrors(err)
	}

	openBankingForwardedUser := models.OpenBankingForwardedUser{
		ConnectorID: connectorID,
		AccessToken: resp.PermanentToken,
		PSPUserID:   resp.PSPUserID,
		Metadata:    resp.Metadata,
	}

	err = e.storage.OpenBankingForwardedUserUpsert(detachedCtx, psuID, openBankingForwardedUser)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	return nil
}

func (e *engine) DeletePaymentServiceUser(ctx context.Context, psuID uuid.UUID) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.DeleteUser")
	defer span.End()

	id := fmt.Sprintf("delete-user-%s-%s", e.stack, psuID.String())
	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference: id,
		},
		Status:    models.TASK_STATUS_PROCESSING,
		CreatedAt: now,
		UpdatedAt: now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	_, err := e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack: e.stack,
			},
		},
		workflow.RunDeletePSU,
		workflow.DeletePSU{
			TaskID: task.ID,
			PsuID:  psuID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	return task, nil
}

func (e *engine) DeletePaymentServiceUserConnector(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.DeleteUserConnector")
	defer span.End()

	id := models.TaskIDReference(fmt.Sprintf("delete-user-connector-%s", e.stack), connectorID, psuID.String())
	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: connectorID,
		},
		ConnectorID: &connectorID,
		Status:      models.TASK_STATUS_PROCESSING,
		CreatedAt:   now,
		UpdatedAt:   now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	_, err := e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connectorID.String(),
			},
		},
		workflow.RunDeletePSUConnector,
		workflow.DeletePSUConnector{
			TaskID:      task.ID,
			ConnectorID: connectorID,
			PsuID:       psuID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	return task, nil
}

func (e *engine) DeletePaymentServiceUserConnection(ctx context.Context, connectorID models.ConnectorID, psuID uuid.UUID, connectionID string) (models.Task, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.DeleteUserConnection")
	defer span.End()

	id := models.TaskIDReference(fmt.Sprintf("delete-user-connection-%s", e.stack), connectorID, psuID.String())
	now := time.Now().UTC()
	task := models.Task{
		ID: models.TaskID{
			Reference:   id,
			ConnectorID: connectorID,
		},
		ConnectorID: &connectorID,
		Status:      models.TASK_STATUS_PROCESSING,
		CreatedAt:   now,
		UpdatedAt:   now,
	}

	if err := e.storage.TasksUpsert(ctx, task); err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	_, err := e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connectorID.String(),
			},
		},
		workflow.RunDeleteConnection,
		workflow.DeleteConnection{
			TaskID:       task.ID,
			ConnectorID:  connectorID,
			PsuID:        psuID,
			ConnectionID: connectionID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return models.Task{}, err
	}

	return task, nil
}

func (e *engine) CreatePaymentServiceUserLink(ctx context.Context, applicationName string, psuID uuid.UUID, connectorID models.ConnectorID, idempotencyKey *uuid.UUID, ClientRedirectURL *string) (string, string, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.CreateUserLink")
	defer span.End()

	plugin, err := e.connectors.Get(connectorID)
	if err != nil {
		otel.RecordError(span, err)
		if errors.Is(err, connectors.ErrNotFound) {
			return "", "", fmt.Errorf("connector %w", ErrNotFound)
		}
		return "", "", err
	}

	psu, err := e.storage.PaymentServiceUsersGet(ctx, psuID)
	if err != nil {
		otel.RecordError(span, err)
		return "", "", err
	}

	openBankingForwardedUser, err := e.storage.OpenBankingForwardedUserGet(ctx, psuID, connectorID)
	if err != nil {
		otel.RecordError(span, err)
		return "", "", err
	}

	id := uuid.New()
	if idempotencyKey != nil {
		id = *idempotencyKey
	}

	attempt := models.OpenBankingConnectionAttempt{
		ID:          id,
		PsuID:       psuID,
		ConnectorID: connectorID,
		CreatedAt:   time.Now().UTC(),
		Status:      models.OpenBankingConnectionAttemptStatusPending,
		State: models.CallbackState{
			Randomized: uuid.New().String(),
			AttemptID:  id,
		},
		ClientRedirectURL: ClientRedirectURL,
	}

	detachedCtx := context.WithoutCancel(ctx)
	e.wg.Add(1)
	defer e.wg.Done()

	err = e.storage.OpenBankingConnectionAttemptsUpsert(
		detachedCtx,
		attempt,
	)
	if err != nil {
		otel.RecordError(span, err)
		return "", "", err
	}

	webhookBaseURL, err := utils.GetWebhookBaseURL(e.stackPublicURL, connectorID)
	if err != nil {
		return "", "", fmt.Errorf("joining webhook base URL: %w", err)
	}

	formanceRedirectURL, err := utils.GetFormanceRedirectURL(e.stackPublicURL, connectorID)
	if err != nil {
		return "", "", fmt.Errorf("joining formance redirect URI: %w", err)
	}

	resp, err := plugin.CreateUserLink(detachedCtx, models.CreateUserLinkRequest{
		ApplicationName:          applicationName,
		AttemptID:                attempt.ID.String(),
		PaymentServiceUser:       models.ToPSPPaymentServiceUser(psu),
		OpenBankingForwardedUser: openBankingForwardedUser,
		ClientRedirectURL:        ClientRedirectURL,
		FormanceRedirectURL:      &formanceRedirectURL,
		CallBackState:            attempt.State.String(),
		WebhookBaseURL:           webhookBaseURL,
	})
	if err != nil {
		otel.RecordError(span, err)
		return "", "", handlePluginErrors(err)
	}

	attempt.TemporaryToken = resp.TemporaryLinkToken
	err = e.storage.OpenBankingConnectionAttemptsUpsert(
		detachedCtx,
		attempt,
	)
	if err != nil {
		return "", "", err
	}

	return attempt.ID.String(), resp.Link, nil
}

func (e *engine) UpdatePaymentServiceUserLink(ctx context.Context, applicationName string, psuID uuid.UUID, connectorID models.ConnectorID, connectionID string, idempotencyKey *uuid.UUID, ClientRedirectURL *string) (string, string, error) {
	ctx, span := otel.Tracer().Start(ctx, "engine.UpdateUserLink")
	defer span.End()

	plugin, err := e.connectors.Get(connectorID)
	if err != nil {
		otel.RecordError(span, err)
		if errors.Is(err, connectors.ErrNotFound) {
			return "", "", fmt.Errorf("connector %w", ErrNotFound)
		}
		return "", "", err
	}

	psu, err := e.storage.PaymentServiceUsersGet(ctx, psuID)
	if err != nil {
		otel.RecordError(span, err)
		return "", "", err
	}

	openBankingForwardedUser, err := e.storage.OpenBankingForwardedUserGet(ctx, psuID, connectorID)
	if err != nil {
		otel.RecordError(span, err)
		return "", "", err
	}

	connection, _, err := e.storage.OpenBankingConnectionsGetFromConnectionID(
		ctx,
		connectorID,
		connectionID,
	)
	if err != nil {
		return "", "", err
	}

	id := uuid.New()
	if idempotencyKey != nil {
		id = *idempotencyKey
	}

	attempt := models.OpenBankingConnectionAttempt{
		ID:          id,
		PsuID:       psuID,
		ConnectorID: connectorID,
		CreatedAt:   time.Now().UTC(),
		Status:      models.OpenBankingConnectionAttemptStatusPending,
		State: models.CallbackState{
			Randomized: uuid.New().String(),
			AttemptID:  id,
		},
		ClientRedirectURL: ClientRedirectURL,
	}

	detachedCtx := context.WithoutCancel(ctx)
	e.wg.Add(1)
	defer e.wg.Done()

	err = e.storage.OpenBankingConnectionAttemptsUpsert(
		detachedCtx,
		attempt,
	)
	if err != nil {
		otel.RecordError(span, err)
		return "", "", err
	}

	webhookBaseURL, err := utils.GetWebhookBaseURL(e.stackPublicURL, connectorID)
	if err != nil {
		return "", "", fmt.Errorf("joining webhook base URL: %w", err)
	}

	formanceRedirectURL, err := utils.GetFormanceRedirectURL(e.stackPublicURL, connectorID)
	if err != nil {
		return "", "", fmt.Errorf("joining formance redirect URI: %w", err)
	}

	resp, err := plugin.UpdateUserLink(detachedCtx, models.UpdateUserLinkRequest{
		AttemptID:                attempt.ID.String(),
		PaymentServiceUser:       models.ToPSPPaymentServiceUser(psu),
		OpenBankingForwardedUser: openBankingForwardedUser,
		Connection:               connection,
		ApplicationName:          applicationName,
		ClientRedirectURL:        ClientRedirectURL,
		FormanceRedirectURL:      &formanceRedirectURL,
		CallBackState:            attempt.State.String(),
		WebhookBaseURL:           webhookBaseURL,
	})
	if err != nil {
		otel.RecordError(span, err)
		return "", "", handlePluginErrors(err)
	}

	attempt.TemporaryToken = resp.TemporaryLinkToken
	err = e.storage.OpenBankingConnectionAttemptsUpsert(
		detachedCtx,
		attempt,
	)
	if err != nil {
		return "", "", err
	}

	return attempt.ID.String(), resp.Link, nil
}

func (e *engine) CompletePaymentServiceUserLink(ctx context.Context, connectorID models.ConnectorID, attemptID uuid.UUID, httpCallInformation models.HTTPCallInformation) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.CompleteUserLink")
	defer span.End()

	ctx = context.WithoutCancel(ctx)
	e.wg.Add(1)
	defer e.wg.Done()

	id := models.TaskIDReference(fmt.Sprintf("complete-user-link-%s", e.stack), connectorID, attemptID.String())
	_, err := e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       id,
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connectorID.String(),
			},
		},
		workflow.RunCompleteUserLink,
		workflow.CompleteUserLink{
			HTTPCallInformation: httpCallInformation,
			ConnectorID:         connectorID,
			AttemptID:           attemptID,
		},
	)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	return nil
}

func (e *engine) HandleWebhook(ctx context.Context, url string, urlPath string, in models.Webhook) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.HandleWebhook")
	defer span.End()

	ctx = context.WithoutCancel(ctx)
	e.wg.Add(1)
	defer e.wg.Done()

	webhooks, config, err := e.verifyAndTrimWebhook(ctx, urlPath, in)
	if err != nil {
		return err
	}

	errGroup, groupCtx := errgroup.WithContext(ctx)
	for _, webhook := range webhooks {
		w := webhook
		errGroup.Go(func() error {
			if _, err := e.temporalClient.ExecuteWorkflow(
				groupCtx,
				client.StartWorkflowOptions{
					ID:                                       fmt.Sprintf("webhook-%s-%s-%s", e.stack, w.ConnectorID.String(), w.ID),
					TaskQueue:                                GetDefaultTaskQueue(e.stack),
					WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
					WorkflowExecutionErrorWhenAlreadyStarted: false,
					SearchAttributes: map[string]interface{}{
						workflow.SearchAttributeStack:       e.stack,
						workflow.SearchAttributeConnectorID: w.ConnectorID.String(),
					},
				},
				workflow.RunHandleWebhooks,
				workflow.HandleWebhooks{
					ConnectorID: w.ConnectorID,
					URL:         url,
					URLPath:     urlPath,
					Webhook:     w,
					Config:      config,
				},
			); err != nil {
				return err
			}
			return nil
		})
	}

	if err := errGroup.Wait(); err != nil {
		otel.RecordError(span, err)
		return err
	}

	return nil
}

func (e *engine) verifyAndTrimWebhook(ctx context.Context, urlPath string, webhook models.Webhook) ([]models.Webhook, *models.WebhookConfig, error) {
	// if the connector has already been uninstalled we want to return a 404 error
	if _, err := e.storage.ConnectorsGet(ctx, webhook.ConnectorID); err != nil {
		return nil, nil, err
	}

	plugin, err := e.connectors.Get(webhook.ConnectorID)
	if err != nil {
		return nil, nil, err
	}

	configs, err := e.storage.WebhooksConfigsGetFromConnectorID(ctx, webhook.ConnectorID)
	if err != nil {
		return nil, nil, err
	}

	var config *models.WebhookConfig
	for _, c := range configs {
		if !strings.Contains(urlPath, c.URLPath) {
			continue
		}

		config = &c
		break
	}

	if config == nil {
		return nil, nil, fmt.Errorf("webhook config %w", ErrNotFound)
	}

	config.FullURL, err = url.JoinPath(e.stackPublicURL, "/api/payments/v3", urlPath)
	if err != nil {
		return nil, nil, err
	}

	verifyResponse, err := plugin.VerifyWebhook(
		ctx,
		models.VerifyWebhookRequest{
			Webhook: models.PSPWebhook{
				BasicAuth:   webhook.BasicAuth,
				QueryValues: webhook.QueryValues,
				Headers:     webhook.Headers,
				Body:        webhook.Body,
			},
			Config: config,
		},
	)
	if err != nil {
		return nil, nil, err
	}

	webhook.IdempotencyKey = verifyResponse.WebhookIdempotencyKey

	webhooks, err := e.trimWebhook(ctx, plugin, config, webhook)
	if err != nil {
		return nil, nil, err
	}

	return webhooks, config, nil
}

func (e *engine) trimWebhook(ctx context.Context, plugin models.Plugin, config *models.WebhookConfig, webhook models.Webhook) ([]models.Webhook, error) {
	trimmedWebhook, err := plugin.TrimWebhook(ctx, models.TrimWebhookRequest{
		Webhook: models.PSPWebhook{
			BasicAuth:   webhook.BasicAuth,
			QueryValues: webhook.QueryValues,
			Headers:     webhook.Headers,
			Body:        webhook.Body,
		},
		Config: config,
	})
	if err != nil {
		return nil, err
	}

	webhooks := make([]models.Webhook, 0, len(trimmedWebhook.Webhooks))
	for i, w := range trimmedWebhook.Webhooks {
		var ik *string
		if webhook.IdempotencyKey != nil {
			ik = pointer.For(fmt.Sprintf("%s-%d", *webhook.IdempotencyKey, i))
		}
		webhooks = append(webhooks, models.Webhook{
			ID:             uuid.New().String(),
			ConnectorID:    webhook.ConnectorID,
			IdempotencyKey: ik,
			BasicAuth:      w.BasicAuth,
			QueryValues:    w.QueryValues,
			Headers:        w.Headers,
			Body:           w.Body,
		})
	}

	return webhooks, nil
}

func (e *engine) CreatePool(ctx context.Context, pool models.Pool) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.CreatePool")
	defer span.End()

	switch pool.Type {
	case models.POOL_TYPE_STATIC:
		eg, groupCtx := errgroup.WithContext(ctx)
		for _, accountID := range pool.PoolAccounts {
			aID := accountID
			eg.Go(func() error {
				acc, err := e.storage.AccountsGet(groupCtx, aID)
				if err != nil {
					return err
				}
				if acc.Type != models.ACCOUNT_TYPE_INTERNAL {
					return fmt.Errorf("account %s is not an internal account: %w", aID, ErrValidation)
				}
				return nil
			})
		}

		if err := eg.Wait(); err != nil {
			otel.RecordError(span, err)
			return err
		}

	case models.POOL_TYPE_DYNAMIC:
		accounts, err := e.populatePoolAccounts(ctx, &pool)
		if err != nil {
			otel.RecordError(span, err)
			return err
		}
		pool.PoolAccounts = accounts
	}

	if err := e.storage.PoolsUpsert(ctx, pool); err != nil {
		otel.RecordError(span, err)
		return err
	}

	// Pool events are now sent via outbox pattern in PoolsUpsert
	return nil
}

func (e *engine) UpdatePoolQuery(ctx context.Context, id uuid.UUID, query map[string]any) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.UpdatePoolQuery")
	defer span.End()

	pool, err := e.storage.PoolsGet(ctx, id)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	if pool.Type == models.POOL_TYPE_STATIC {
		return fmt.Errorf("pool %s is a static pool, use the pool account addition or deletion endpoints: %w", id, ErrValidation)
	}

	pool.Query = query
	// populate the pool accounts before sending the event.
	pool.PoolAccounts, err = e.populatePoolAccounts(ctx, pool)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	if err := e.storage.PoolsUpdateQuery(ctx, *pool, query); err != nil {
		otel.RecordError(span, err)
		return err
	}

	return nil
}
func (e *engine) AddAccountToPool(ctx context.Context, id uuid.UUID, accountID models.AccountID) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.AddAccountToPool")
	defer span.End()

	// Check if the account exists and if it's an INTERNAL account
	account, err := e.storage.AccountsGet(ctx, accountID)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	if account.Type != models.ACCOUNT_TYPE_INTERNAL {
		return fmt.Errorf("account %s is not an internal account: %w", accountID, ErrValidation)
	}

	// Check next if the pool is dynamic, in that case, we send an error to the
	// user telling him to use the pool query update endpoint.
	pool, err := e.storage.PoolsGet(ctx, id)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	if pool.Type == models.POOL_TYPE_DYNAMIC {
		return fmt.Errorf("pool %s is a dynamic pool, use the pool query update endpoint to add an account: %w", id, ErrValidation)
	}

	if err := e.storage.PoolsAddAccount(ctx, id, accountID); err != nil {
		otel.RecordError(span, err)
		return err
	}

	// Pool events are now sent via outbox pattern in PoolsUpsert
	return nil
}

func (e *engine) RemoveAccountFromPool(ctx context.Context, id uuid.UUID, accountID models.AccountID) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.RemoveAccountFromPool")
	defer span.End()

	pool, err := e.storage.PoolsGet(ctx, id)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	if pool.Type == models.POOL_TYPE_DYNAMIC {
		return fmt.Errorf("pool %s is a dynamic pool, use the pool query update endpoint to remove an account: %w", id, ErrValidation)
	}

	if err := e.storage.PoolsRemoveAccount(ctx, id, accountID); err != nil {
		otel.RecordError(span, err)
		return err
	}

	// Pool events are now sent via outbox pattern in PoolsUpsert
	return nil
}

func (e *engine) DeletePool(ctx context.Context, poolID uuid.UUID) error {
	ctx, span := otel.Tracer().Start(ctx, "engine.DeletePool")
	defer span.End()
	_, err := e.storage.PoolsDelete(ctx, poolID)
	if err != nil {
		otel.RecordError(span, err)
		return err
	}

	// Pool deletion events are now sent via outbox pattern in PoolsDelete
	return nil
}

func (e *engine) populatePoolAccounts(ctx context.Context, pool *models.Pool) ([]models.AccountID, error) {
	queryJSON, err := json.Marshal(pool.Query)
	if err != nil {
		return nil, fmt.Errorf("cannot marshal pool query: %w", err)
	}

	qb, err := query.ParseJSON(string(queryJSON))
	if err != nil {
		return nil, fmt.Errorf("cannot parse pool query: %w", err)
	}

	q := storage.NewListAccountsQuery(
		paginate.NewPaginatedQueryOptions(storage.AccountQuery{}).
			WithPageSize(100).
			WithQueryBuilder(qb),
	)

	res := make([]models.AccountID, 0)
	for {
		cursor, err := e.storage.AccountsList(ctx, q)
		if err != nil {
			return nil, fmt.Errorf("cannot list accounts: %w", err)
		}

		for _, account := range cursor.Data {
			res = append(res, account.ID)
		}

		if !cursor.HasMore {
			break
		}

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		if err != nil {
			return nil, fmt.Errorf("cannot unmarshal cursor: %w", err)
		}
	}

	return res, nil
}

func (e *engine) OnStop(ctx context.Context) {
	waitingChan := make(chan struct{})
	go func() {
		e.wg.Wait()
		close(waitingChan)
	}()

	select {
	case <-waitingChan:
	case <-ctx.Done():
	}
}

func (e *engine) OnStart(ctx context.Context) error {
	if err := e.storage.ListenConnectorsChanges(ctx, storage.HandlerConnectorsChanges{
		storage.ConnectorChangesInsert: e.onInsertPlugin,
		storage.ConnectorChangesUpdate: e.onUpdatePlugin,
		storage.ConnectorChangesDelete: e.onDeletePlugin,
	}); err != nil {
		return fmt.Errorf("failed to start engine connector changes listener: %w", err)
	}

	query := storage.NewListConnectorsQuery(
		paginate.NewPaginatedQueryOptions(storage.ConnectorQuery{}).
			WithPageSize(100),
	)

	for {
		connectors, err := e.storage.ConnectorsList(ctx, query)
		if err != nil {
			return err
		}

		for _, connector := range connectors.Data {
			if err := e.onStartPlugin(ctx, connector); err != nil {
				return err
			}
		}

		if !connectors.HasMore {
			break
		}

		err = paginate.UnmarshalCursor(connectors.Next, &query)
		if err != nil {
			return err
		}
	}
	return nil
}

func (e *engine) onInsertPlugin(ctx context.Context, connectorID models.ConnectorID) error {
	e.logger.Debugf("api got insert notification for %q", connectorID.String())
	connector, err := e.storage.ConnectorsGet(ctx, connectorID)
	if err != nil {
		return err
	}

	// skip strict polling period validation if installed by another instance
	_, _, err = e.connectors.Load(*connector, false, false)
	if err != nil {
		return err
	}

	// No need to launch the install workflow here

	return nil
}

func (e *engine) onUpdatePlugin(ctx context.Context, connectorID models.ConnectorID) error {
	e.logger.Debugf("api got update notification for %q", connectorID.String())
	connector, err := e.storage.ConnectorsGet(ctx, connectorID)
	if err != nil {
		if errors.Is(err, storage.ErrNotFound) {
			return e.onDeletePlugin(ctx, connectorID)
		}
		return err
	}

	// skip strict polling period validation if installed by another instance
	_, _, err = e.connectors.Load(*connector, true, false)
	if err != nil {
		e.logger.Errorf("failed to register plugin after update to connector %q: %v", connector.ID.String(), err)
		return err
	}
	return nil
}

func (e *engine) onDeletePlugin(ctx context.Context, connectorID models.ConnectorID) error {
	e.logger.Debugf("api got delete notification for %q", connectorID.String())
	e.connectors.Unload(connectorID)
	return nil
}

func (e *engine) onStartPlugin(ctx context.Context, connector models.Connector) error {
	// skip strict polling period validation if installed by another instance
	if _, _, err := e.connectors.Load(connector, false, false); err != nil {
		// We don't want to crash the server if the plugin registration fails,
		// otherwise, the client will not be able to remove the failing
		// connector from the database because of the crashes.
		// We just log the error and continue.
		e.logger.Errorf("failed to register plugin for connector %q: %s", connector.ID.String(), err.Error())
		return nil
	}

	return nil
}

func (e *engine) launchInstallWorkflow(ctx context.Context, connector models.Connector, config models.Config) (client.WorkflowRun, error) {
	return e.temporalClient.ExecuteWorkflow(
		ctx,
		client.StartWorkflowOptions{
			ID:                                       e.taskIDReferenceFor(IDPrefixConnectorInstall, connector.ID, ""),
			TaskQueue:                                GetDefaultTaskQueue(e.stack),
			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
			WorkflowExecutionErrorWhenAlreadyStarted: false,
			SearchAttributes: map[string]interface{}{
				workflow.SearchAttributeStack:       e.stack,
				workflow.SearchAttributeConnectorID: connector.ID.String(),
			},
		},
		workflow.RunInstallConnector,
		workflow.InstallConnector{
			ConnectorID: connector.ID,
		},
	)
}

var _ Engine = &engine{}