activityImpl.go

package interpreter

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"slices"
	"time"

	"github.com/indeedeng/iwf/service/common/blobstore"

	"github.com/indeedeng/iwf/gen/iwfidl"
	"github.com/indeedeng/iwf/service"
	"github.com/indeedeng/iwf/service/common/compatibility"
	"github.com/indeedeng/iwf/service/common/event"
	"github.com/indeedeng/iwf/service/common/log"
	"github.com/indeedeng/iwf/service/common/ptr"
	"github.com/indeedeng/iwf/service/common/rpc"
	"github.com/indeedeng/iwf/service/common/urlautofix"
	"github.com/indeedeng/iwf/service/interpreter/env"
	"github.com/indeedeng/iwf/service/interpreter/interfaces"
)

// StateStart is Deprecated, will be removed in next release
func StateStart(
	ctx context.Context, backendType service.BackendType, input service.StateStartActivityInput, searchAttributes []iwfidl.SearchAttribute,
) (*iwfidl.WorkflowStateStartResponse, error) {
	return StateApiWaitUntil(ctx, backendType, input, searchAttributes)
}

func StateApiWaitUntil(
	ctx context.Context, backendType service.BackendType, input service.StateStartActivityInput, searchAttributes []iwfidl.SearchAttribute,
) (*iwfidl.WorkflowStateStartResponse, error) {
	stateApiWaitUntilStartTime := time.Now().UnixMilli()
	provider := interfaces.GetActivityProviderByType(backendType)
	logger := provider.GetLogger(ctx)
	logger.Info("StateWaitUntilActivity", "input", log.ToJsonAndTruncateForLogging(input))
	iwfWorkerBaseUrl := urlautofix.FixWorkerUrl(input.IwfWorkerUrl)

	svcCfg := env.GetSharedConfig()
	apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
		DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
		Servers: []iwfidl.ServerConfiguration{
			{
				URL: iwfWorkerBaseUrl,
			},
		},
	})

	activityInfo := provider.GetActivityInfo(ctx)
	attempt := activityInfo.Attempt
	scheduledTs := activityInfo.ScheduledTime.Unix()
	input.Request.Context.Attempt = &attempt
	input.Request.Context.FirstAttemptTimestamp = &scheduledTs

	var err error
	if input.Request.StateInput != nil && input.Request.StateInput.ExtStoreId != nil {
		_, err = loadStateInputFromExternalStorage(ctx, input.Request.StateInput)
		if err != nil {
			if activityInfo.IsLocalActivity {
				reqBytes, _ := json.Marshal(input.Request)
				logger.Warn("StateApiWaitUntil local activity return on error",
					"workflowId", activityInfo.WorkflowExecution.ID,
					"stateExecutionId", input.Request.Context.GetStateExecutionId(),
					"payloadSize", len(reqBytes))
			}
			return nil, err
		}
	}

	// Load data attributes from external storage
	err = blobstore.LoadDataObjectsFromExternalStorage(ctx, input.Request.DataObjects, env.GetBlobStore())
	if err != nil {
		if activityInfo.IsLocalActivity {
			reqBytes, _ := json.Marshal(input.Request)
			logger.Warn("StateApiWaitUntil local activity return on error",
				"workflowId", activityInfo.WorkflowExecution.ID,
				"stateExecutionId", input.Request.Context.GetStateExecutionId(),
				"payloadSize", len(reqBytes))
		}
		return nil, err
	}

	req := apiClient.DefaultApi.ApiV1WorkflowStateStartPost(ctx)
	resp, httpResp, err := req.WorkflowStateStartRequest(input.Request).Execute()
	printDebugMsg(logger, err, iwfWorkerBaseUrl)
	if checkHttpError(err, httpResp) {
		stateStartErr := composeHttpError(
			activityInfo.IsLocalActivity,
			provider, err, httpResp, string(iwfidl.STATE_API_FAIL_ERROR_TYPE))
		errType, errDetails := env.GetUnifiedClient().GetApplicationErrorTypeAndDetails(stateStartErr)

		event.Handle(iwfidl.IwfEvent{
			EventType:          iwfidl.STATE_WAIT_UNTIL_ATTEMPT_FAIL_EVENT,
			WorkflowType:       input.Request.WorkflowType,
			WorkflowId:         activityInfo.WorkflowExecution.ID,
			WorkflowRunId:      activityInfo.WorkflowExecution.RunID,
			StateId:            ptr.Any(input.Request.WorkflowStateId),
			StateExecutionId:   ptr.Any(input.Request.Context.GetStateExecutionId()),
			StartTimestampInMs: ptr.Any(stateApiWaitUntilStartTime),
			EndTimestampInMs:   ptr.Any(time.Now().UnixMilli()),
			SearchAttributes:   searchAttributes,
			Error: &iwfidl.IwfEventError{
				Type:    &errType,
				Details: &errDetails,
			},
		})
		if activityInfo.IsLocalActivity {
			reqBytes, _ := json.Marshal(input.Request)
			logger.Warn("StateApiWaitUntil local activity return on error",
				"workflowId", activityInfo.WorkflowExecution.ID,
				"stateExecutionId", input.Request.Context.GetStateExecutionId(),
				"payloadSize", len(reqBytes))
		}
		return nil, stateStartErr
	}

	if err := checkCommandRequestFromWaitUntilResponse(resp); err != nil {
		stateStartErr := composeStartApiRespError(provider, err, resp)
		errType, errDetails := env.GetUnifiedClient().GetApplicationErrorTypeAndDetails(stateStartErr)

		event.Handle(iwfidl.IwfEvent{
			EventType:          iwfidl.STATE_WAIT_UNTIL_ATTEMPT_FAIL_EVENT,
			WorkflowType:       input.Request.WorkflowType,
			WorkflowId:         activityInfo.WorkflowExecution.ID,
			WorkflowRunId:      activityInfo.WorkflowExecution.RunID,
			StateId:            ptr.Any(input.Request.WorkflowStateId),
			StateExecutionId:   ptr.Any(input.Request.Context.GetStateExecutionId()),
			StartTimestampInMs: ptr.Any(stateApiWaitUntilStartTime),
			EndTimestampInMs:   ptr.Any(time.Now().UnixMilli()),
			SearchAttributes:   searchAttributes,
			Error: &iwfidl.IwfEventError{
				Type:    &errType,
				Details: &errDetails,
			},
		})
		if activityInfo.IsLocalActivity {
			reqBytes, _ := json.Marshal(input.Request)
			logger.Warn("StateApiWaitUntil local activity return on error",
				"workflowId", activityInfo.WorkflowExecution.ID,
				"stateExecutionId", input.Request.Context.GetStateExecutionId(),
				"payloadSize", len(reqBytes))
		}
		return nil, stateStartErr
	}

	// Before returning successful results, check if it's local activity then compose some info for debug purpose
	// This is because local activity doesn't record input into the history.
	// But there are some small info that are important to record
	if activityInfo.IsLocalActivity {
		resp.LocalActivityInput = composeInputForDebug(input.Request.Context.GetStateExecutionId())
	}

	if env.GetSharedConfig().ExternalStorage.Enabled && env.GetBlobStore() != nil {
		err = blobstore.WriteDataObjectsToExternalStorage(ctx, resp.UpsertDataObjects, activityInfo.WorkflowExecution.ID, env.GetSharedConfig().ExternalStorage.ThresholdInBytes, env.GetBlobStore(), env.GetSharedConfig().ExternalStorage.Enabled)
		if err != nil {
			if activityInfo.IsLocalActivity {
				reqBytes, _ := json.Marshal(input.Request)
				logger.Warn("StateApiWaitUntil local activity return on error",
					"workflowId", activityInfo.WorkflowExecution.ID,
					"stateExecutionId", input.Request.Context.GetStateExecutionId(),
					"payloadSize", len(reqBytes))
			}
			return nil, err
		}
	}

	event.Handle(iwfidl.IwfEvent{
		EventType:          iwfidl.STATE_WAIT_UNTIL_ATTEMPT_SUCC_EVENT,
		WorkflowType:       input.Request.WorkflowType,
		WorkflowId:         activityInfo.WorkflowExecution.ID,
		WorkflowRunId:      activityInfo.WorkflowExecution.RunID,
		StateId:            ptr.Any(input.Request.WorkflowStateId),
		StateExecutionId:   ptr.Any(input.Request.Context.GetStateExecutionId()),
		StartTimestampInMs: ptr.Any(stateApiWaitUntilStartTime),
		EndTimestampInMs:   ptr.Any(time.Now().UnixMilli()),
		SearchAttributes:   searchAttributes,
	})
	if activityInfo.IsLocalActivity {
		respBytes, _ := json.Marshal(resp)
		logger.Warn("StateApiWaitUntil local activity return on success",
			"workflowId", activityInfo.WorkflowExecution.ID,
			"stateExecutionId", input.Request.Context.GetStateExecutionId(),
			"payloadSize", len(respBytes))
	}
	return resp, nil
}

// StateDecide is deprecated. Will be removed in next release
func StateDecide(
	ctx context.Context,
	backendType service.BackendType,
	input service.StateDecideActivityInput,
) (*iwfidl.WorkflowStateDecideResponse, error) {
	return StateApiExecute(ctx, backendType, input)
}

func StateApiExecute(
	ctx context.Context,
	backendType service.BackendType,
	input service.StateDecideActivityInput,
) (*iwfidl.WorkflowStateDecideResponse, error) {
	stateApiExecuteStartTime := time.Now().UnixMilli()
	provider := interfaces.GetActivityProviderByType(backendType)
	logger := provider.GetLogger(ctx)
	logger.Info("StateExecuteActivity", "input", log.ToJsonAndTruncateForLogging(input))

	iwfWorkerBaseUrl := urlautofix.FixWorkerUrl(input.IwfWorkerUrl)
	svcCfg := env.GetSharedConfig()
	apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
		DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
		Servers: []iwfidl.ServerConfiguration{
			{
				URL: iwfWorkerBaseUrl,
			},
		},
	})

	activityInfo := provider.GetActivityInfo(ctx)
	attempt := activityInfo.Attempt
	scheduledTs := activityInfo.ScheduledTime.Unix()
	input.Request.Context.Attempt = &attempt
	input.Request.Context.FirstAttemptTimestamp = &scheduledTs

	var wholeStateInputCopy *iwfidl.EncodedObject
	var err error
	if input.Request.StateInput != nil && input.Request.StateInput.ExtStoreId != nil {
		wholeStateInputCopy, err = loadStateInputFromExternalStorage(ctx, input.Request.StateInput)
		if err != nil {
			if activityInfo.IsLocalActivity {
				reqBytes, _ := json.Marshal(input.Request)
				logger.Warn("StateApiExecute local activity return on error",
					"workflowId", activityInfo.WorkflowExecution.ID,
					"stateExecutionId", input.Request.Context.GetStateExecutionId(),
					"payloadSize", len(reqBytes))
			}
			return nil, err
		}
	}

	// Load data attributes from external storage
	err = blobstore.LoadDataObjectsFromExternalStorage(ctx, input.Request.DataObjects, env.GetBlobStore())
	if err != nil {
		if activityInfo.IsLocalActivity {
			reqBytes, _ := json.Marshal(input.Request)
			logger.Warn("StateApiExecute local activity return on error",
				"workflowId", activityInfo.WorkflowExecution.ID,
				"stateExecutionId", input.Request.Context.GetStateExecutionId(),
				"payloadSize", len(reqBytes))
		}
		return nil, err
	}

	// workflowImpl is only passing StateWaitUntilFailed, not StateStartApiSucceeded, to save the history from storing duplicate data
	// So we need to construct the other for backward compatibility(to old SDK that is still using StateStartApiSucceeded)
	if input.Request.CommandResults != nil {
		if input.Request.CommandResults.StateWaitUntilFailed == nil {
			input.Request.CommandResults.StateStartApiSucceeded = iwfidl.PtrBool(true)
			input.Request.CommandResults.StateWaitUntilFailed = iwfidl.PtrBool(false)
		} else {
			input.Request.CommandResults.StateStartApiSucceeded = iwfidl.PtrBool(!*input.Request.CommandResults.StateWaitUntilFailed)
		}
	}

	req := apiClient.DefaultApi.ApiV1WorkflowStateDecidePost(ctx)
	resp, httpResp, err := req.WorkflowStateDecideRequest(input.Request).Execute()
	printDebugMsg(logger, err, iwfWorkerBaseUrl)
	if checkHttpError(err, httpResp) {
		stateApiExecuteErr := composeHttpError(
			activityInfo.IsLocalActivity,
			provider, err, httpResp, string(iwfidl.STATE_API_FAIL_ERROR_TYPE))

		errType, errDetails := env.GetUnifiedClient().GetApplicationErrorTypeAndDetails(stateApiExecuteErr)

		event.Handle(iwfidl.IwfEvent{
			EventType:          iwfidl.STATE_EXECUTE_ATTEMPT_FAIL_EVENT,
			WorkflowType:       input.Request.WorkflowType,
			WorkflowId:         activityInfo.WorkflowExecution.ID,
			WorkflowRunId:      activityInfo.WorkflowExecution.RunID,
			StateId:            ptr.Any(input.Request.WorkflowStateId),
			StateExecutionId:   input.Request.Context.StateExecutionId,
			StartTimestampInMs: ptr.Any(stateApiExecuteStartTime),
			EndTimestampInMs:   ptr.Any(time.Now().UnixMilli()),
			SearchAttributes:   input.Request.SearchAttributes,
			Error: &iwfidl.IwfEventError{
				Type:    &errType,
				Details: &errDetails,
			},
		})
		if activityInfo.IsLocalActivity {
			reqBytes, _ := json.Marshal(input.Request)
			logger.Warn("StateApiExecute local activity return on error",
				"workflowId", activityInfo.WorkflowExecution.ID,
				"stateExecutionId", input.Request.Context.GetStateExecutionId(),
				"payloadSize", len(reqBytes))
		}
		return nil, stateApiExecuteErr
	}

	if err = checkStateDecisionFromResponse(resp); err != nil {
		stateApiExecuteErr := composeExecuteApiRespError(provider, err, resp)
		errType, errDetails := env.GetUnifiedClient().GetApplicationErrorTypeAndDetails(stateApiExecuteErr)

		event.Handle(iwfidl.IwfEvent{
			EventType:          iwfidl.STATE_EXECUTE_ATTEMPT_FAIL_EVENT,
			WorkflowType:       input.Request.WorkflowType,
			WorkflowId:         activityInfo.WorkflowExecution.ID,
			WorkflowRunId:      activityInfo.WorkflowExecution.RunID,
			StateId:            ptr.Any(input.Request.WorkflowStateId),
			StateExecutionId:   input.Request.Context.StateExecutionId,
			StartTimestampInMs: ptr.Any(stateApiExecuteStartTime),
			EndTimestampInMs:   ptr.Any(time.Now().UnixMilli()),
			SearchAttributes:   input.Request.SearchAttributes,
			Error: &iwfidl.IwfEventError{
				Type:    &errType,
				Details: &errDetails,
			},
		})
		if activityInfo.IsLocalActivity {
			reqBytes, _ := json.Marshal(input.Request)
			logger.Warn("StateApiExecute local activity return on error",
				"workflowId", activityInfo.WorkflowExecution.ID,
				"stateExecutionId", input.Request.Context.GetStateExecutionId(),
				"payloadSize", len(reqBytes))
		}
		return nil, stateApiExecuteErr
	}

	// Before returning successful results, check if it's local activity then compose some info for debug purpose
	// This is because local activity doesn't record input into the history.
	// But there are some small info that are important to record
	if activityInfo.IsLocalActivity {
		resp.LocalActivityInput = composeInputForDebug(input.Request.Context.GetStateExecutionId())
	}

	// Externalize only when enabled and blob store is available (nil when e.g. STAGING_LEVEL was empty at worker start).
	if env.GetSharedConfig().ExternalStorage.Enabled && env.GetBlobStore() != nil {
		resp.StateDecision.NextStates, err = writeNextStateInputsToExternalStorage(ctx, resp.StateDecision.NextStates, wholeStateInputCopy, activityInfo.WorkflowExecution.ID)
		if err != nil {
			if activityInfo.IsLocalActivity {
				reqBytes, _ := json.Marshal(input.Request)
				logger.Warn("StateApiExecute local activity return on error",
					"workflowId", activityInfo.WorkflowExecution.ID,
					"stateExecutionId", input.Request.Context.GetStateExecutionId(),
					"payloadSize", len(reqBytes))
			}
			return nil, err
		}
		err = blobstore.WriteDataObjectsToExternalStorage(ctx, resp.UpsertDataObjects, activityInfo.WorkflowExecution.ID, env.GetSharedConfig().ExternalStorage.ThresholdInBytes, env.GetBlobStore(), env.GetSharedConfig().ExternalStorage.Enabled)
		if err != nil {
			if activityInfo.IsLocalActivity {
				reqBytes, _ := json.Marshal(input.Request)
				logger.Warn("StateApiExecute local activity return on error",
					"workflowId", activityInfo.WorkflowExecution.ID,
					"stateExecutionId", input.Request.Context.GetStateExecutionId(),
					"payloadSize", len(reqBytes))
			}
			return nil, err
		}
	}

	event.Handle(iwfidl.IwfEvent{
		EventType:          iwfidl.STATE_EXECUTE_ATTEMPT_SUCC_EVENT,
		WorkflowType:       input.Request.WorkflowType,
		WorkflowId:         activityInfo.WorkflowExecution.ID,
		WorkflowRunId:      activityInfo.WorkflowExecution.RunID,
		StateId:            ptr.Any(input.Request.WorkflowStateId),
		StateExecutionId:   input.Request.Context.StateExecutionId,
		StartTimestampInMs: ptr.Any(stateApiExecuteStartTime),
		EndTimestampInMs:   ptr.Any(time.Now().UnixMilli()),
		SearchAttributes:   input.Request.SearchAttributes,
	})
	if activityInfo.IsLocalActivity {
		respBytes, _ := json.Marshal(resp)
		logger.Warn("StateApiExecute local activity return on success",
			"workflowId", activityInfo.WorkflowExecution.ID,
			"stateExecutionId", input.Request.Context.GetStateExecutionId(),
			"payloadSize", len(respBytes))
	}
	return resp, nil
}

func composeInputForDebug(stateExeId string) *string {
	// NOTE: only use the stateExecutionId for now, but we can add more later if needed
	return iwfidl.PtrString(fmt.Sprintf("stateExeId: %s", stateExeId))
}

func checkStateDecisionFromResponse(resp *iwfidl.WorkflowStateDecideResponse) error {
	if resp == nil || resp.StateDecision == nil || len(resp.StateDecision.NextStates) == 0 {
		return fmt.Errorf("empty state decision is no longer supported. If it's from old SDKs then upgrade the SDK to newer versions")
	}
	return nil
}

func printDebugMsg(logger interfaces.UnifiedLogger, err error, url string) {
	debugMode := os.Getenv(service.EnvNameDebugMode)
	if debugMode != "" {
		logger.Info("check error at http request", err, url)
	}
}

func composeStartApiRespError(provider interfaces.ActivityProvider, err error, resp *iwfidl.WorkflowStateStartResponse) error {
	respStr, _ := resp.MarshalJSON()
	return provider.NewApplicationError(string(iwfidl.STATE_API_FAIL_ERROR_TYPE),
		fmt.Sprintf("err msg: %v, response: %v", err, string(respStr)))
}

func composeExecuteApiRespError(provider interfaces.ActivityProvider, err error, resp *iwfidl.WorkflowStateDecideResponse) error {
	respStr, _ := resp.MarshalJSON()
	return provider.NewApplicationError(string(iwfidl.STATE_API_FAIL_ERROR_TYPE),
		fmt.Sprintf("err msg: %v, response: %v", err, string(respStr)))
}

func checkHttpError(err error, httpResp *http.Response) bool {
	if err != nil || (httpResp != nil && httpResp.StatusCode != http.StatusOK) {
		return true
	}
	return false
}

func composeHttpError(
	isLocalActivity bool, provider interfaces.ActivityProvider, err error, httpResp *http.Response, errType string,
) error {
	responseBody := "None"
	var statusCode int
	if httpResp != nil {
		body, err := io.ReadAll(httpResp.Body)
		if err != nil {
			responseBody = "cannot read body from http response"
		} else {
			responseBody = string(body)
		}
		statusCode = httpResp.StatusCode
	}
	errMsg := err.Error()
	var trimmedResponseBody, trimmedErrMsg string
	if isLocalActivity {
		trimmedErrMsg = trimText(errMsg, 5)
		trimmedResponseBody = trimText(responseBody, 50)
		errType = "1st-attempt-failure"
	} else {
		trimmedErrMsg = trimText(errMsg, 50)
		trimmedResponseBody = trimText(responseBody, 500)
	}

	return provider.NewApplicationError(errType,
		fmt.Sprintf("statusCode: %v, responseBody: %v, errMsg: %v", statusCode, trimmedResponseBody, trimmedErrMsg))
}

func trimText(msg string, maxLength int) string {
	if len(msg) > maxLength {
		return msg[:maxLength] + "..."
	}
	return msg
}

func checkCommandRequestFromWaitUntilResponse(resp *iwfidl.WorkflowStateStartResponse) error {
	if resp == nil || resp.CommandRequest == nil {
		return nil
	}
	commandReq := resp.CommandRequest
	if len(commandReq.GetTimerCommands())+len(commandReq.GetSignalCommands())+len(commandReq.GetInterStateChannelCommands()) > 0 {
		dtt := compatibility.GetDeciderTriggerType(*commandReq)
		if dtt != iwfidl.ANY_COMMAND_COMPLETED && dtt != iwfidl.ALL_COMMAND_COMPLETED && dtt != iwfidl.ANY_COMMAND_COMBINATION_COMPLETED {
			return fmt.Errorf("unsupported decider trigger type %s", dtt)
		}
		if dtt == iwfidl.ANY_COMMAND_COMBINATION_COMPLETED {
			// every command must have an id for this type
			err := fmt.Errorf("ANY_COMMAND_COMBINATION_COMPLETED can only be used when every command has an commandId, and the combination list cannot be empty")
			if len(commandReq.GetCommandCombinations()) == 0 {
				return err
			}
			for _, cmd := range commandReq.GetTimerCommands() {
				if cmd.GetCommandId() == "" {
					return err
				}
			}
			for _, cmd := range commandReq.GetSignalCommands() {
				if cmd.GetCommandId() == "" {
					return err
				}
			}
			for _, cmd := range commandReq.GetInterStateChannelCommands() {
				if cmd.GetCommandId() == "" {
					return err
				}
			}
			// Check if each command in the combinations has a matching command in one of the lists
			if !areAllCommandCombinationsIdsValid(commandReq) {
				return fmt.Errorf("ANY_COMMAND_COMBINATION_COMPLETED can only be used when every command has an commandId that is found in TimerCommands, SignalCommands or InternalChannelCommand")
			}
		}
	}
	// NOTE: we don't require decider trigger type when there is no commands
	return nil
}

func areAllCommandCombinationsIdsValid(commandReq *iwfidl.CommandRequest) bool {
	timerSignalInternalChannelCmdIds := listTimerSignalInternalChannelCommandIds(commandReq)
	for _, commandCombo := range commandReq.GetCommandCombinations() {
		for _, cmdId := range commandCombo.GetCommandIds() {
			if !slices.Contains(timerSignalInternalChannelCmdIds, cmdId) {
				return false
			}
		}
	}
	return true
}

func listTimerSignalInternalChannelCommandIds(commandReq *iwfidl.CommandRequest) []string {
	var ids []string
	for _, timerCmd := range commandReq.GetTimerCommands() {
		ids = append(ids, timerCmd.GetCommandId())
	}
	for _, signalCmd := range commandReq.GetSignalCommands() {
		ids = append(ids, signalCmd.GetCommandId())
	}
	for _, internalChannelCmd := range commandReq.GetInterStateChannelCommands() {
		ids = append(ids, internalChannelCmd.GetCommandId())
	}
	return ids
}

func DumpWorkflowInternal(
	ctx context.Context, backendType service.BackendType, req iwfidl.WorkflowDumpRequest,
) (*iwfidl.WorkflowDumpResponse, error) {
	provider := interfaces.GetActivityProviderByType(backendType)
	logger := provider.GetLogger(ctx)
	logger.Info("DumpWorkflowInternalActivity", "input", log.ToJsonAndTruncateForLogging(req))

	svcCfg := env.GetSharedConfig()
	apiAddress := svcCfg.GetApiServiceAddressWithDefault()

	apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
		DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
		Servers: []iwfidl.ServerConfiguration{
			{
				URL: apiAddress,
			},
		},
	})

	request := apiClient.DefaultApi.ApiV1WorkflowInternalDumpPost(ctx)
	resp, httpResp, err := request.WorkflowDumpRequest(req).Execute()
	if checkHttpError(err, httpResp) {
		return nil, composeHttpError(provider.GetActivityInfo(ctx).IsLocalActivity,
			provider, err, httpResp, string(iwfidl.SERVER_INTERNAL_ERROR_TYPE))
	}
	return resp, nil
}

func InvokeWorkerRpc(
	ctx context.Context, backendType service.BackendType, rpcPrep *service.PrepareRpcQueryResponse,
	req iwfidl.WorkflowRpcRequest,
) (*interfaces.InvokeRpcActivityOutput, error) {
	provider := interfaces.GetActivityProviderByType(backendType)
	logger := provider.GetLogger(ctx)
	logger.Info("InvokeWorkerRpcActivity", "input", log.ToJsonAndTruncateForLogging(req))
	activityInfo := provider.GetActivityInfo(ctx)

	apiMaxSeconds := env.GetSharedConfig().Api.MaxWaitSeconds

	resp, statusErr := rpc.InvokeWorkerRpc(ctx, rpcPrep, req, apiMaxSeconds, env.GetBlobStore(), env.GetSharedConfig().ExternalStorage)
	output := &interfaces.InvokeRpcActivityOutput{
		RpcOutput:   resp,
		StatusError: statusErr,
	}
	if activityInfo.IsLocalActivity {
		outputBytes, _ := json.Marshal(output)
		logger.Warn("InvokeWorkerRpc local activity return",
			"workflowId", activityInfo.WorkflowExecution.ID,
			"payloadSize", len(outputBytes))
	}
	return output, nil
}

func writeNextStateInputsToExternalStorage(ctx context.Context, nextStates []iwfidl.StateMovement, currentInputCopy *iwfidl.EncodedObject, workflowId string) ([]iwfidl.StateMovement, error) {
	for i := range nextStates {
		if err := processNextStateInputForExternalStorage(ctx, nextStates[i].StateInput, currentInputCopy, workflowId); err != nil {
			return nil, err
		}
	}
	return nextStates, nil
}

func processNextStateInputForExternalStorage(ctx context.Context, nextStateInput *iwfidl.EncodedObject, currentInputCopy *iwfidl.EncodedObject, workflowId string) error {
	blobStore := env.GetBlobStore()

	// Check if external storage is needed
	if nextStateInput == nil || nextStateInput.Data == nil || len(*nextStateInput.Data) <= env.GetSharedConfig().ExternalStorage.ThresholdInBytes {
		return nil
	}

	// Try to reuse existing external storage if data is identical
	if currentInputCopy != nil && currentInputCopy.Data != nil && nextStateInput.Data != nil &&
		*currentInputCopy.Data == *nextStateInput.Data &&
		currentInputCopy.ExtStoreId != nil && currentInputCopy.ExtPath != nil {
		// Reuse existing external storage
		nextStateInput.ExtStoreId = currentInputCopy.ExtStoreId
		nextStateInput.ExtPath = currentInputCopy.ExtPath
		nextStateInput.Data = nil
		return nil
	}

	// Save to external storage
	storeId, path, err := blobStore.WriteObject(ctx, workflowId, *nextStateInput.Data)
	if err != nil {
		return err
	}
	nextStateInput.ExtStoreId = &storeId
	nextStateInput.ExtPath = &path
	nextStateInput.Data = nil
	return nil
}

// loadStateInputFromExternalStorage is specifically for loading state input from external storage.
// It loads the data and replace the field of input,
// and it also preserves the external storage identifiers for potential reuse optimization by returning a "whole" encodedObject
func loadStateInputFromExternalStorage(ctx context.Context, input *iwfidl.EncodedObject) (*iwfidl.EncodedObject, error) {
	if input == nil || input.ExtStoreId == nil || input.ExtPath == nil {
		return input, nil
	}

	extStoreId := *input.ExtStoreId
	extPath := *input.ExtPath

	err := doLoadFromExternalStorage(ctx, input)
	if err != nil {
		return nil, err
	}

	// Load the data but preserve external storage identifiers for potential reuse
	// This allows us to optimize by reusing the same external storage location
	// if the next state input has identical data
	newEncodedObject := iwfidl.EncodedObject{
		Data:       input.Data,
		Encoding:   input.Encoding,
		ExtStoreId: &extStoreId,
		ExtPath:    &extPath,
	}
	return &newEncodedObject, nil
}

// doLoadFromExternalStorage loads data from external storage and replace the fields of the input
func doLoadFromExternalStorage(ctx context.Context, input *iwfidl.EncodedObject) error {
	blobStore := env.GetBlobStore()

	data, err := blobStore.ReadObject(ctx, input.GetExtStoreId(), input.GetExtPath())
	if err != nil {
		return err
	}

	input.Data = &data
	input.ExtPath = nil
	input.ExtStoreId = nil
	return nil
}

func CleanupBlobStore(
	ctx context.Context, backendType service.BackendType, storeId string,
) (int, error) {
	store := env.GetBlobStore()
	provider := interfaces.GetActivityProviderByType(backendType)
	logger := provider.GetLogger(ctx)
	logger.Info("CleanupBlobStore started")

	client := env.GetUnifiedClient()

	var continueToken *string
	var totalDeleted int

	for {
		listOutput, err := store.ListWorkflowPaths(ctx, blobstore.ListObjectPathsInput{
			StoreId:           storeId,
			ContinuationToken: continueToken,
		})
		if err != nil {
			return totalDeleted, err
		}
		continueToken = listOutput.ContinuationToken
		for _, workflowPath := range listOutput.WorkflowPaths {
			_, valid := blobstore.ExtractYyyymmddToUnixSeconds(workflowPath)
			if !valid {
				logger.Info("CleanupBlobStore skipped workflow path", "path", workflowPath)
				continue
			}

			// Check if workflow still exists in Temporal
			// We must always check because workflows can run indefinitely,
			// and the retention period only applies after workflow closure
			workflowId := blobstore.MustExtractWorkflowId(workflowPath)
			_, err := client.DescribeWorkflowExecution(ctx, workflowId, "", nil)
			if client.IsNotFoundError(err) {
				// Workflow has been removed from Temporal, safe to delete S3 objects
				err = store.DeleteWorkflowObjects(ctx, storeId, workflowPath)
				if err != nil {
					logger.Error("CleanupBlobStore failed to delete workflow objects", "workflowPath", workflowPath, "error", err)
					return totalDeleted, err
				}
				totalDeleted++
				logger.Info("CleanupBlobStore deleted workflow objects", "workflowPath", workflowPath)
			} else if err != nil {
				logger.Error("CleanupBlobStore failed to describe workflow", "workflowPath", workflowPath, "error", err)
				return totalDeleted, err
			}
			// If no error, workflow still exists (open or within retention), don't delete

			// this is a long running activity
			// using record heartbeat so that it won't timeout at startToClose timeout
			provider.RecordHeartbeat(ctx)
		}
		if continueToken == nil {
			break
		}
	}
	logger.Info("CleanupBlobStore completed", "totalDeleted", totalDeleted)
	return totalDeleted, nil
}