validator.go

package activity

import (
	"github.com/google/uuid"
	activitypb "go.temporal.io/api/activity/v1"
	commonpb "go.temporal.io/api/common/v1"
	enumspb "go.temporal.io/api/enums/v1"
	"go.temporal.io/api/serviceerror"
	"go.temporal.io/api/workflowservice/v1"
	"go.temporal.io/server/common"
	"go.temporal.io/server/common/dynamicconfig"
	"go.temporal.io/server/common/log"
	"go.temporal.io/server/common/log/tag"
	"go.temporal.io/server/common/namespace"
	"go.temporal.io/server/common/primitives/timestamp"
	"go.temporal.io/server/common/priorities"
	"go.temporal.io/server/common/retrypolicy"
	"go.temporal.io/server/common/searchattribute"
	"go.temporal.io/server/common/tqid"
	"google.golang.org/protobuf/types/known/durationpb"
)

// ValidateAndNormalizeStandaloneActivity validates and normalizes the attributes for a standalone activity.
func ValidateAndNormalizeStandaloneActivity(
	activityID string,
	activityType string,
	getDefaultActivityRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings],
	maxIDLengthLimit int,
	namespaceID namespace.ID,
	options *activitypb.ActivityOptions,
	priority *commonpb.Priority,
	runTimeout *durationpb.Duration,
) error {
	// Standalone activities always use user defined task queues, so we can enforce user defined task queue validation
	if err := tqid.NormalizeAndValidateUserDefined(options.TaskQueue, "", "", maxIDLengthLimit); err != nil {
		return err
	}

	return validateAndNormalizeActivityAttributes(
		activityID,
		activityType,
		getDefaultActivityRetrySettings,
		maxIDLengthLimit,
		namespaceID,
		options,
		priority,
		runTimeout)
}

// ValidateAndNormalizeEmbeddedActivity validates and normalizes the attributes for an embedded activity.
func ValidateAndNormalizeEmbeddedActivity(
	activityID string,
	activityType string,
	getDefaultActivityRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings],
	maxIDLengthLimit int,
	namespaceID namespace.ID,
	options *activitypb.ActivityOptions,
	priority *commonpb.Priority,
	runTimeout *durationpb.Duration,
	workflowTaskQueueName string,
) error {
	if err := tqid.NormalizeAndValidateUserDefined(options.TaskQueue, "", workflowTaskQueueName, maxIDLengthLimit); err != nil {
		return err
	}

	return validateAndNormalizeActivityAttributes(
		activityID,
		activityType,
		getDefaultActivityRetrySettings,
		maxIDLengthLimit,
		namespaceID,
		options,
		priority,
		runTimeout)
}

// ValidateAndNormalizeActivityAttributes validates and normalizes the common activity request attributes.
// This validation is shared by both standalone and embedded activities.
// IMPORTANT: this method mutates the input params; in cases where it's critical to maintain immutability
// (i.e., when incoming request can potentially be retried), clone the params first before passing it in.
//
// The timeout normalization logic is as follows:
// 1. If ScheduleToClose is set, fill in missing ScheduleToStart and StartToClose from ScheduleToClose
// 2. If StartToClose is set but ScheduleToClose is not set, set ScheduleToClose to runTimeout, and fill in missing ScheduleToStart from runTimeout
// 3. If neither ScheduleToClose nor StartToClose is set, return error
// 4. Ensure all timeouts do not exceed runTimeout if runTimeout is set (>0)
// 5. Ensure HeartbeatTimeout does not exceed StartToClose
func validateAndNormalizeActivityAttributes(
	activityID string,
	activityType string,
	getDefaultActivityRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings],
	maxIDLengthLimit int,
	namespaceID namespace.ID,
	options *activitypb.ActivityOptions,
	priority *commonpb.Priority,
	runTimeout *durationpb.Duration,
) error {
	if activityID == "" {
		return serviceerror.NewInvalidArgument("activityId is not set")
	}
	if activityType == "" {
		return serviceerror.NewInvalidArgument("activityType is not set")
	}

	if err := validateActivityRetryPolicy(namespaceID, options.RetryPolicy, getDefaultActivityRetrySettings); err != nil {
		return err
	}

	if len(activityID) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("activityId exceeds length limit. Length=%d Limit=%d",
			len(activityID), maxIDLengthLimit)
	}
	if len(activityType) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("activityType exceeds length limit. Length=%d Limit=%d",
			len(activityType), maxIDLengthLimit)
	}

	if err := priorities.Validate(priority); err != nil {
		return serviceerror.NewInvalidArgumentf("invalid priorities: %v", err)
	}

	return validateAndNormalizeTimeouts(activityID,
		activityType,
		runTimeout,
		options)
}

func validateStartDelay(startDelay *durationpb.Duration) error {
	if err := timestamp.ValidateAndCapProtoDuration(startDelay); err != nil {
		return serviceerror.NewInvalidArgumentf("invalid StartDelay: %v", err)
	}
	return nil
}

func validateActivityRetryPolicy(
	namespaceID namespace.ID,
	retryPolicy *commonpb.RetryPolicy,
	getDefaultActivityRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings],
) error {
	if retryPolicy == nil {
		return nil
	}
	// TODO(saa-preview): this is a namespace setting, not a namespace id setting
	defaultActivityRetrySettings := getDefaultActivityRetrySettings(namespaceID.String())
	retrypolicy.EnsureDefaults(retryPolicy, defaultActivityRetrySettings)
	return retrypolicy.Validate(retryPolicy)
}

func validateAndNormalizeTimeouts(
	activityID string,
	activityType string,
	runTimeout *durationpb.Duration,
	options *activitypb.ActivityOptions,
) error {
	// Only attempt to deduce and fill in unspecified timeouts only when all timeouts are non-negative.
	if err := timestamp.ValidateAndCapProtoDuration(options.GetScheduleToCloseTimeout()); err != nil {
		return serviceerror.NewInvalidArgumentf("invalid ScheduleToCloseTimeout: %v", err)
	}
	if err := timestamp.ValidateAndCapProtoDuration(options.GetScheduleToStartTimeout()); err != nil {
		return serviceerror.NewInvalidArgumentf("invalid ScheduleToStartTimeout: %v", err)
	}
	if err := timestamp.ValidateAndCapProtoDuration(options.GetStartToCloseTimeout()); err != nil {
		return serviceerror.NewInvalidArgumentf("invalid StartToCloseTimeout: %v", err)
	}
	if err := timestamp.ValidateAndCapProtoDuration(options.GetHeartbeatTimeout()); err != nil {
		return serviceerror.NewInvalidArgumentf("invalid HeartbeatTimeout: %v", err)
	}

	scheduleToCloseSet := options.GetScheduleToCloseTimeout().AsDuration() > 0
	scheduleToStartSet := options.GetScheduleToStartTimeout().AsDuration() > 0
	startToCloseSet := options.GetStartToCloseTimeout().AsDuration() > 0

	if scheduleToCloseSet {
		if scheduleToStartSet {
			options.ScheduleToStartTimeout = timestamp.MinDurationPtr(options.ScheduleToStartTimeout, options.ScheduleToCloseTimeout)
		} else {
			options.ScheduleToStartTimeout = options.ScheduleToCloseTimeout
		}
		if startToCloseSet {
			options.StartToCloseTimeout = timestamp.MinDurationPtr(options.StartToCloseTimeout, options.ScheduleToCloseTimeout)
		} else {
			options.StartToCloseTimeout = options.ScheduleToCloseTimeout
		}
	} else if startToCloseSet {
		// We are in !validScheduleToClose due to the first if above
		options.ScheduleToCloseTimeout = runTimeout
		if !scheduleToStartSet {
			options.ScheduleToStartTimeout = runTimeout
		}
	} else {
		// Deduction failed as there's not enough information to fill in missing timeouts.
		return serviceerror.NewInvalidArgumentf("a valid StartToClose or ScheduleToCloseTimeout is not set on ScheduleActivityTaskCommand. ActivityId=%s ActivityType=%s",
			activityID, activityType)
	}
	// ensure activity timeout never larger than workflow timeout
	if runTimeout.AsDuration() > 0 {
		runTimeoutDur := runTimeout.AsDuration()
		if options.ScheduleToCloseTimeout.AsDuration() > runTimeoutDur {
			options.ScheduleToCloseTimeout = runTimeout
		}
		if options.ScheduleToStartTimeout.AsDuration() > runTimeoutDur {
			options.ScheduleToStartTimeout = runTimeout
		}
		if options.StartToCloseTimeout.AsDuration() > runTimeoutDur {
			options.StartToCloseTimeout = runTimeout
		}
		if options.HeartbeatTimeout.AsDuration() > runTimeoutDur {
			options.HeartbeatTimeout = runTimeout
		}
	}

	options.HeartbeatTimeout = timestamp.MinDurationPtr(options.HeartbeatTimeout, options.StartToCloseTimeout)

	return nil
}

func validateAndNormalizeIDPolicy(req *workflowservice.StartActivityExecutionRequest) error {
	if req.GetIdReusePolicy() == enumspb.ACTIVITY_ID_REUSE_POLICY_UNSPECIFIED {
		req.IdReusePolicy = enumspb.ACTIVITY_ID_REUSE_POLICY_ALLOW_DUPLICATE
	}

	if req.GetIdConflictPolicy() == enumspb.ACTIVITY_ID_CONFLICT_POLICY_UNSPECIFIED {
		req.IdConflictPolicy = enumspb.ACTIVITY_ID_CONFLICT_POLICY_FAIL
	}

	return nil
}

// validateOnConflictOptions validates the on_conflict_options of a start request:
//   - attach_completion_callbacks requires attach_request_id. A completion callback is recorded
//     against the request ID (see addCompletionCallbacks, which keys the callback by request ID).
//   - attach_request_id requires at least one completion callback or link, since attaching a
//     request ID is only meaningful alongside something to attach.
//
// attach_links is independent and may be set on its own.
func validateOnConflictOptions(req *workflowservice.StartActivityExecutionRequest) error {
	onConflictOptions := req.GetOnConflictOptions()
	if onConflictOptions == nil {
		return nil
	}
	if onConflictOptions.GetAttachCompletionCallbacks() && !onConflictOptions.GetAttachRequestId() {
		return serviceerror.NewInvalidArgument(
			"on_conflict_options: attach_completion_callbacks requires attach_request_id to be set")
	}
	if onConflictOptions.GetAttachRequestId() &&
		len(req.GetCompletionCallbacks()) == 0 &&
		len(req.GetLinks()) == 0 {
		return serviceerror.NewInvalidArgument(
			"on_conflict_options: attach_request_id requires at least one completion callback or link")
	}
	return nil
}

func validateBlobSize(
	activityID string,
	blobSizeViolationTagValue string,
	blobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter,
	blobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter,
	blobSize int,
	logger log.Logger,
	namespaceName string,
) error {
	sizeWarnLimit := blobSizeLimitWarn(namespaceName)
	sizeErrorLimit := blobSizeLimitError(namespaceName)

	if blobSize > sizeWarnLimit {
		logger.Warn("Activity blob size exceeds the warning limit.",
			tag.WorkflowNamespace(namespaceName),
			tag.ActivityID(activityID),
			tag.ActivitySize(int64(blobSize)),
			tag.BlobSizeViolationOperation(blobSizeViolationTagValue))
	}

	if blobSize > sizeErrorLimit {
		return common.ErrBlobSizeExceedsLimit
	}

	return nil
}

func validateAndNormalizeSearchAttributes(
	req *workflowservice.StartActivityExecutionRequest,
	saMapperProvider searchattribute.MapperProvider,
	saValidator *searchattribute.Validator,
) error {
	namespaceName := req.GetNamespace()

	// Unalias search attributes for validation.
	saToValidate := req.SearchAttributes
	if saMapperProvider != nil && saToValidate != nil {
		var err error
		saToValidate, err = searchattribute.UnaliasFields(saMapperProvider, saToValidate, namespaceName)
		if err != nil {
			return err
		}
	}

	if err := saValidator.Validate(saToValidate, namespaceName); err != nil {
		return err
	}

	return saValidator.ValidateSize(saToValidate, namespaceName)
}

func validateDescribeActivityExecutionRequest(
	req *workflowservice.DescribeActivityExecutionRequest,
	maxIDLengthLimit int,
) error {
	if req.GetActivityId() == "" {
		return serviceerror.NewInvalidArgument("activity ID is required")
	}
	if len(req.GetActivityId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("activity ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetActivityId()), maxIDLengthLimit)
	}
	hasRunID := req.GetRunId() != ""
	hasLongPollToken := len(req.GetLongPollToken()) > 0

	if hasLongPollToken && !hasRunID {
		return serviceerror.NewInvalidArgument("run id is required when long poll token is provided")
	}
	if hasRunID {
		_, err := uuid.Parse(req.GetRunId())
		if err != nil {
			return serviceerror.NewInvalidArgument("invalid run id: must be a valid UUID")
		}
	}
	return nil
}

func validatePollActivityExecutionRequest(
	req *workflowservice.PollActivityExecutionRequest,
	maxIDLengthLimit int,
) error {
	if req.GetActivityId() == "" {
		return serviceerror.NewInvalidArgument("activity ID is required")
	}
	if len(req.GetActivityId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("activity ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetActivityId()), maxIDLengthLimit)
	}
	if runID := req.GetRunId(); runID != "" {
		_, err := uuid.Parse(runID)
		if err != nil {
			return serviceerror.NewInvalidArgument("invalid run id: must be a valid UUID")
		}
	}
	return nil
}

func validateAndNormalizeStartRequest(
	req *workflowservice.StartActivityExecutionRequest,
	maxIDLengthLimit int,
	blobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter,
	blobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter,
	logger log.Logger,
	saMapperProvider searchattribute.MapperProvider,
	saValidator *searchattribute.Validator,
) error {
	if req.GetRequestId() == "" {
		req.RequestId = uuid.NewString()
	} else if len(req.GetRequestId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("request ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetRequestId()), maxIDLengthLimit)
	}

	if len(req.GetIdentity()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("identity exceeds length limit. Length=%d Limit=%d",
			len(req.GetIdentity()), maxIDLengthLimit)
	}

	if err := validateAndNormalizeIDPolicy(req); err != nil {
		return err
	}

	if err := validateBlobSize(
		req.GetActivityId(),
		"StartActivityExecution",
		blobSizeLimitError,
		blobSizeLimitWarn,
		req.Input.Size(),
		logger,
		req.GetNamespace()); err != nil {
		return serviceerror.NewInvalidArgument("input exceeds length limit")
	}

	if req.GetSearchAttributes() != nil {
		if err := validateAndNormalizeSearchAttributes(
			req,
			saMapperProvider,
			saValidator); err != nil {
			return err
		}
	}

	return nil
}

func validateAndNormalizeCancelRequest(
	req *workflowservice.RequestCancelActivityExecutionRequest,
	maxIDLengthLimit int,
	blobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter,
	blobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter,
	logger log.Logger,
) error {
	if req.GetActivityId() == "" {
		return serviceerror.NewInvalidArgument("activity ID is required")
	}

	if len(req.GetActivityId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("activity ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetActivityId()), maxIDLengthLimit)
	}

	if req.GetRequestId() == "" {
		req.RequestId = uuid.NewString()
	} else if len(req.GetRequestId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("request ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetRequestId()), maxIDLengthLimit)
	}

	if len(req.GetIdentity()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("identity exceeds length limit. Length=%d Limit=%d",
			len(req.GetIdentity()), maxIDLengthLimit)
	}

	if runID := req.GetRunId(); runID != "" {
		_, err := uuid.Parse(runID)
		if err != nil {
			return serviceerror.NewInvalidArgument("invalid run id: must be a valid UUID")
		}
	}

	err := validateBlobSize(
		req.GetActivityId(),
		"RequestCancelActivityExecution",
		blobSizeLimitError,
		blobSizeLimitWarn,
		len(req.GetReason()),
		logger,
		req.GetNamespace())
	if err != nil {
		return serviceerror.NewInvalidArgument("reason exceeds length limit")
	}

	return nil
}

func validateAndNormalizeDeleteRequest(
	req *workflowservice.DeleteActivityExecutionRequest,
	maxIDLengthLimit int,
) error {
	if req.GetActivityId() == "" {
		return serviceerror.NewInvalidArgument("activity ID is required")
	}

	if len(req.GetActivityId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("activity ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetActivityId()), maxIDLengthLimit)
	}

	if runID := req.GetRunId(); runID != "" {
		_, err := uuid.Parse(runID)
		if err != nil {
			return serviceerror.NewInvalidArgument("invalid run id: must be a valid UUID")
		}
	}

	return nil
}

func validateAndNormalizeTerminateRequest(
	req *workflowservice.TerminateActivityExecutionRequest,
	maxIDLengthLimit int,
	blobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter,
	blobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter,
	logger log.Logger,
) error {
	if req.GetActivityId() == "" {
		return serviceerror.NewInvalidArgument("activity ID is required")
	}

	if len(req.GetActivityId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("activity ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetActivityId()), maxIDLengthLimit)
	}

	if req.GetRequestId() == "" {
		req.RequestId = uuid.NewString()
	} else if len(req.GetRequestId()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("request ID exceeds length limit. Length=%d Limit=%d",
			len(req.GetRequestId()), maxIDLengthLimit)
	}

	if len(req.GetIdentity()) > maxIDLengthLimit {
		return serviceerror.NewInvalidArgumentf("identity exceeds length limit. Length=%d Limit=%d",
			len(req.GetIdentity()), maxIDLengthLimit)
	}

	if runID := req.GetRunId(); runID != "" {
		_, err := uuid.Parse(runID)
		if err != nil {
			return serviceerror.NewInvalidArgument("invalid run id: must be a valid UUID")
		}
	}

	err := validateBlobSize(
		req.GetActivityId(),
		"TerminateActivityExecution",
		blobSizeLimitError,
		blobSizeLimitWarn,
		len(req.GetReason()),
		logger,
		req.GetNamespace())
	if err != nil {
		return serviceerror.NewInvalidArgument("reason exceeds length limit")
	}

	return nil
}