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

package workflow

import (
	"encoding/json"
	"time"

	"github.com/hanzo-fi/go-libs/v5/pkg/observe/log"
	temporalworker "github.com/hanzo-fi/go-libs/v5/pkg/workflow/temporal"
	"github.com/hanzo-fi/payments/internal/connectors"
	"go.temporal.io/sdk/client"
	"go.temporal.io/sdk/workflow"
)

const (
	SearchAttributeScheduleID  = "PaymentScheduleID"
	SearchAttributeStack       = "Stack"
	SearchAttributeConnectorID = "ConnectorID"
)

type FromPayload struct {
	ID      string          `json:"id"`
	Payload json.RawMessage `json:"payload"`
}

func (f *FromPayload) GetPayload() json.RawMessage {
	if f == nil {
		return nil
	}
	return f.Payload
}

type Workflow struct {
	temporalNamespace string
	temporalClient    client.Client

	connectors connectors.Manager

	stackPublicURL      string
	stack               string
	healthCheckInterval time.Duration

	logger logging.Logger
}

func New(temporalClient client.Client, temporalNamespace string, connectors connectors.Manager, stack string, stackPublicURL string, logger logging.Logger, healthCheckInterval time.Duration) Workflow {
	return Workflow{
		temporalClient:      temporalClient,
		temporalNamespace:   temporalNamespace,
		connectors:          connectors,
		stack:               stack,
		stackPublicURL:      stackPublicURL,
		healthCheckInterval: healthCheckInterval,
		logger:              logger,
	}
}

func (w Workflow) DefinitionSet() temporalworker.DefinitionSet {
	return temporalworker.NewDefinitionSet().
		Append(temporalworker.Definition{
			Name: RunFetchNextAccounts,
			Func: w.runFetchNextAccounts,
		}).
		Append(temporalworker.Definition{
			Name: RunFetchNextBalances,
			Func: w.runFetchNextBalances,
		}).
		Append(temporalworker.Definition{
			Name: RunFetchNextExternalAccounts,
			Func: w.runFetchNextExternalAccounts,
		}).
		Append(temporalworker.Definition{
			Name: RunFetchNextOthers,
			Func: w.runFetchNextOthers,
		}).
		Append(temporalworker.Definition{
			Name: RunFetchNextPayments,
			Func: w.runFetchNextPayments,
		}).
		Append(temporalworker.Definition{
			Name: RunFetchNextOrders,
			Func: w.runFetchNextOrders,
		}).
		Append(temporalworker.Definition{
			Name: RunFetchNextConversions,
			Func: w.runFetchNextConversions,
		}).
		Append(temporalworker.Definition{
			Name: RunListActiveSchedules,
			Func: w.runListActiveSchedules,
		}).
		Append(temporalworker.Definition{
			Name: RunUpdateSchedulePollingPeriod,
			Func: w.runUpdateSchedulePollingPeriod,
		}).
		Append(temporalworker.Definition{
			Name: RunTerminateScheduleByID,
			Func: w.runTerminateScheduleByID,
		}).
		Append(temporalworker.Definition{
			Name: RunTerminateSchedules,
			Func: w.runTerminateSchedules,
		}).
		Append(temporalworker.Definition{
			Name: RunTerminateWorkflows,
			Func: w.runTerminateWorkflows,
		}).
		Append(temporalworker.Definition{
			Name: RunInstallConnector,
			Func: w.runInstallConnector,
		}).
		Append(temporalworker.Definition{
			Name: RunBootstrapTasks,
			Func: w.runBootstrapTasks,
		}).
		Append(temporalworker.Definition{
			Name: RunBootstrapTask,
			Func: w.runBootstrapTask,
		}).
		Append(temporalworker.Definition{
			Name: RunResetConnector,
			Func: w.runResetConnector,
		}).
		Append(temporalworker.Definition{
			Name: RunUninstallConnector,
			Func: w.runUninstallConnector,
		}).
		Append(temporalworker.Definition{
			Name: RunCreateBankAccount,
			Func: w.runCreateBankAccount,
		}).
		Append(temporalworker.Definition{
			Name: RunCreatePayout,
			Func: w.runCreatePayout,
		}).
		Append(temporalworker.Definition{
			Name: RunReversePayout,
			Func: w.runReversePayout,
		}).
		Append(temporalworker.Definition{
			Name: RunPollPayout,
			Func: w.runPollPayout,
		}).
		Append(temporalworker.Definition{
			Name: RunCreateTransfer,
			Func: w.runCreateTransfer,
		}).
		Append(temporalworker.Definition{
			Name: RunReverseTransfer,
			Func: w.runReverseTransfer,
		}).
		Append(temporalworker.Definition{
			Name: RunPollTransfer,
			Func: w.runPollTransfer,
		}).
		Append(temporalworker.Definition{
			Name: RunNextTasksV3_1,
			Func: w.runNextTasksV3_1,
		}).
		Append(temporalworker.Definition{
			Name: RunCreateWebhooks,
			Func: w.runCreateWebhooks,
		}).
		Append(temporalworker.Definition{
			Name: RunHandleWebhooks,
			Func: w.runHandleWebhooks,
		}).
		Append(temporalworker.Definition{
			Name: RunStoreWebhookTranslation,
			Func: w.runStoreWebhookTranslation,
		}).
		Append(temporalworker.Definition{
			Name: RunDeletePSU,
			Func: w.runDeletePSU,
		}).
		Append(temporalworker.Definition{
			Name: RunDeletePSUConnector,
			Func: w.runDeletePSUConnector,
		}).
		Append(temporalworker.Definition{
			Name: RunDeleteConnection,
			Func: w.runDeleteConnection,
		}).
		Append(temporalworker.Definition{
			Name: RunCompleteUserLink,
			Func: w.runCompleteUserLink,
		}).
		Append(temporalworker.Definition{
			Name: RunFetchOpenBankingData,
			Func: w.runFetchOpenBankingData,
		}).
		Append(temporalworker.Definition{
			Name: RunDeleteOpenBankingConnectionData,
			Func: w.runDeleteOpenBankingConnectionData,
		}).
		Append(temporalworker.Definition{
			Name: RunOutboxPublisher,
			Func: w.runOutboxPublisher,
		}).
		Append(temporalworker.Definition{
			Name: RunOutboxCleanup,
			Func: w.runOutboxCleanup,
		}).
		Append(temporalworker.Definition{
			Name: RunConnectorHealthCheck,
			Func: w.runConnectorHealthCheck,
		}).
		Append(temporalworker.Definition{
			Name: RunScheduleConnectorHealthCheck,
			Func: w.runScheduleConnectorHealthCheck,
		}).
		Append(temporalworker.Definition{
			Name: RunNextTasks,   //nolint:staticcheck
			Func: w.runNextTasks, //nolint:staticcheck
		}).
		Append(temporalworker.Definition{
			Name: RunSendEvents,   //nolint:staticcheck // ignore deprecation
			Func: w.runSendEvents, //nolint:staticcheck // ignore deprecation
		}).
		Append(temporalworker.Definition{
			Name: RunUpdatePaymentInitiationFromPayment,   //nolint:staticcheck // ignore deprecation
			Func: w.runUpdatePaymentInitiationFromPayment, //nolint:staticcheck // ignore deprecation
		})
}

func (w Workflow) shouldContinueAsNew(ctx workflow.Context) bool {
	workflowInfo := workflow.GetInfo(ctx)
	return workflowInfo.GetContinueAsNewSuggested()
}