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

package workflow

import (
	"fmt"
	"time"

	"github.com/hanzo-fi/payments/internal/connectors/engine/activities"
	"github.com/hanzo-fi/payments/pkg/domain/models"
	"github.com/pkg/errors"
	"go.temporal.io/api/enums/v1"
	"go.temporal.io/sdk/client"
	"go.temporal.io/sdk/workflow"
)

func (w Workflow) runNextTasksV3_1(
	ctx workflow.Context,
	connectorID models.ConnectorID,
	fromPayload *FromPayload,
	taskTree []models.ConnectorTaskTree,
) error {
	var nextWorkflow interface{}
	var request interface{}
	var capability models.Capability

	for _, task := range taskTree {
		switch task.TaskType {
		case models.TASK_FETCH_ACCOUNTS:
			req := FetchNextAccounts{
				ConnectorID:  connectorID,
				FromPayload:  fromPayload,
				Periodically: task.Periodically,
			}

			nextWorkflow = RunFetchNextAccounts
			request = req
			capability = models.CAPABILITY_FETCH_ACCOUNTS

		case models.TASK_FETCH_EXTERNAL_ACCOUNTS:
			req := FetchNextExternalAccounts{
				ConnectorID:  connectorID,
				FromPayload:  fromPayload,
				Periodically: task.Periodically,
			}

			nextWorkflow = RunFetchNextExternalAccounts
			request = req
			capability = models.CAPABILITY_FETCH_EXTERNAL_ACCOUNTS

		case models.TASK_FETCH_OTHERS:
			req := FetchNextOthers{
				ConnectorID:  connectorID,
				Name:         task.Name,
				FromPayload:  fromPayload,
				Periodically: task.Periodically,
			}

			nextWorkflow = RunFetchNextOthers
			request = req
			capability = models.CAPABILITY_FETCH_OTHERS

		case models.TASK_FETCH_PAYMENTS:
			req := FetchNextPayments{
				ConnectorID:  connectorID,
				FromPayload:  fromPayload,
				Periodically: task.Periodically,
			}

			nextWorkflow = RunFetchNextPayments
			request = req
			capability = models.CAPABILITY_FETCH_PAYMENTS

		case models.TASK_FETCH_BALANCES:
			req := FetchNextBalances{
				ConnectorID:  connectorID,
				FromPayload:  fromPayload,
				Periodically: task.Periodically,
			}

			nextWorkflow = RunFetchNextBalances
			request = req
			capability = models.CAPABILITY_FETCH_BALANCES

		case models.TASK_FETCH_ORDERS:
			req := FetchNextOrders{
				ConnectorID:  connectorID,
				FromPayload:  fromPayload,
				Periodically: task.Periodically,
			}

			nextWorkflow = RunFetchNextOrders
			request = req
			capability = models.CAPABILITY_FETCH_ORDERS

		case models.TASK_FETCH_CONVERSIONS:
			req := FetchNextConversions{
				ConnectorID:  connectorID,
				FromPayload:  fromPayload,
				Periodically: task.Periodically,
			}

			nextWorkflow = RunFetchNextConversions
			request = req
			capability = models.CAPABILITY_FETCH_CONVERSIONS

		case models.TASK_CREATE_WEBHOOKS:
			req := CreateWebhooks{
				ConnectorID: connectorID,
				FromPayload: fromPayload,
			}

			nextWorkflow = RunCreateWebhooks
			request = req
			capability = models.CAPABILITY_CREATE_WEBHOOKS

		default:
			return fmt.Errorf("unknown task type: %v", task.TaskType)
		}

		// Schedule next workflow every polling duration
		if task.Periodically {
			// TODO(polo): context
			err := w.scheduleNextWorkflow(
				ctx,
				connectorID,
				capability,
				task,
				fromPayload,
				nextWorkflow,
				request,
			)
			if err != nil {
				return fmt.Errorf("failed to schedule periodic task: %w", err)
			}

			continue
		}

		// Run next workflow immediately
		if err := workflow.ExecuteChildWorkflow(
			workflow.WithChildOptions(
				ctx,
				workflow.ChildWorkflowOptions{
					TaskQueue:             w.getDefaultTaskQueue(),
					ParentClosePolicy:     enums.PARENT_CLOSE_POLICY_ABANDON,
					SearchAttributes:      w.SearchAttributes(ctx, &connectorID),
				},
			),
			nextWorkflow,
			request,
			task.NextTasks,
		).GetChildWorkflowExecution().Get(ctx, nil); err != nil {
			return errors.Wrap(err, "running next workflow")
		}
	}

	return nil
}

func fetchNextWorkflowScheduleID(stack, connectorID, capability string, fromPayload *FromPayload) string {
	if fromPayload == nil {
		return fmt.Sprintf("%s-%s-%s", stack, connectorID, capability)
	}
	return fmt.Sprintf("%s-%s-%s-%s", stack, connectorID, capability, fromPayload.ID)
}

func (w Workflow) scheduleNextWorkflow(
	ctx workflow.Context,
	connectorID models.ConnectorID,
	capability models.Capability,
	task models.ConnectorTaskTree,
	fromPayload *FromPayload,
	nextWorkflow interface{},
	request interface{},
) error {
	scheduleID := fetchNextWorkflowScheduleID(w.stack, connectorID.String(), capability.String(), fromPayload)

	err := activities.StorageSchedulesStore(
		infiniteRetryContext(ctx),
		models.Schedule{
			ID:          scheduleID,
			ConnectorID: connectorID,
			CreatedAt:   workflow.Now(ctx).UTC(),
		})
	if err != nil {
		return err
	}

	pollingPeriod, err := w.connectorPollingPeriod(ctx, connectorID)
	if err != nil {
		return err
	}

	err = activities.TemporalScheduleCreate(
		infiniteRetryContext(ctx),
		activities.ScheduleCreateOptions{
			ScheduleID: scheduleID,
			Jitter:     calculateJitter(pollingPeriod),
			Interval: &client.ScheduleIntervalSpec{
				Every: pollingPeriod,
			},
			Action: client.ScheduleWorkflowAction{
				// Use the same ID as the schedule ID, so we can identify the workflows running.
				// This is useful for debugging purposes.
				ID:       scheduleID,
				Workflow: nextWorkflow,
				Args: []interface{}{
					request,
					task.NextTasks,
				},
				TaskQueue: w.getDefaultTaskQueue(),
			},
			// schedules are recreated at application start to ensure that any changes made to a connector's workflow take effect
			// allow the workflow from the outdated schedule to finish running before starting a new one with the new workflow
			Overlap:            enums.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE,
			TriggerImmediately: true,
			SearchAttributes:   w.ScheduleSearchAttributes(ctx, &connectorID, scheduleID),
		},
	)
	if err != nil {
		return err
	}
	return nil
}

func calculateJitter(pollingPeriod time.Duration) time.Duration {
	maxJitter := time.Minute * 5
	jitter := pollingPeriod / 2
	if jitter <= maxJitter {
		return jitter
	}
	return maxJitter
}

// Deprecated: use RunNextTasksV3_1 instead, we keep that it during 3.0 => 3.1 migration
func (w Workflow) runNextTasks(
	ctx workflow.Context,
	config models.Config,
	connectorID models.ConnectorID,
	fromPayload *FromPayload,
	taskTree []models.ConnectorTaskTree,
) error {
	connector, err := activities.StorageConnectorsGet(infiniteRetryContext(ctx), connectorID)
	if err != nil {
		return err
	}

	// avoid scheduling next workflow if connector has been flagged for deletion
	if connector.ScheduledForDeletion {
		return nil
	}

	return w.runNextTasksV3_1(
		ctx,
		connectorID,
		fromPayload,
		taskTree,
	)
}

func (w Workflow) runNextTaskAsChildWorkflow(ctx workflow.Context, connectorID models.ConnectorID, nextTasks []models.ConnectorTaskTree, wg workflow.WaitGroup, fromPayload *FromPayload, errChan chan<- error) {
	wg.Add(1)
	workflow.Go(ctx, func(ctx workflow.Context) {
		defer wg.Done()

		if err := workflow.ExecuteChildWorkflow(
			workflow.WithChildOptions(
				ctx,
				workflow.ChildWorkflowOptions{
					TaskQueue:             w.getDefaultTaskQueue(),
					ParentClosePolicy:     enums.PARENT_CLOSE_POLICY_ABANDON,
					SearchAttributes:      w.SearchAttributes(ctx, &connectorID),
				},
			),
			RunNextTasks, //nolint:staticcheck // ignore deprecated
			models.Config{},
			connectorID,
			fromPayload,
			nextTasks,
		).Get(ctx, nil); err != nil {
			errChan <- errors.Wrap(err, "running next workflow")
		}
	})
}

const RunNextTasksV3_1 = "RunNextTasksV3_1"

// Deprecated: use RunNextTasksV3_1 instead, we keep that it during 3.0 => 3.1 migration
const RunNextTasks = "Run" //nolint:staticcheck