nexusoperation

var (
	Archetype   = chasm.FullyQualifiedName(libraryName, componentName)
	ArchetypeID = chasm.GenerateTypeID(Archetype)
)

var (
	EndpointSearchAttribute  = chasm.NewSearchAttributeKeyword("Endpoint", chasm.SearchAttributeFieldKeyword01)
	ServiceSearchAttribute   = chasm.NewSearchAttributeKeyword("Service", chasm.SearchAttributeFieldKeyword02)
	OperationSearchAttribute = chasm.NewSearchAttributeKeyword("Operation", chasm.SearchAttributeFieldKeyword03)
	RequestIDSearchAttribute = chasm.NewSearchAttributeKeyword("RequestId", chasm.SearchAttributeFieldKeyword04)
	StatusSearchAttribute    = chasm.NewSearchAttributeKeyword("ExecutionStatus", chasm.SearchAttributeFieldLowCardinalityKeyword01)
)

var (
	ErrResponseBodyTooLarge  = errors.New("http: response body too large")
	ErrInvalidOperationToken = errors.New("invalid operation token")
)

var CallbackURLTemplate = dynamicconfig.NewGlobalTypedSettingWithConverter(
	"nexusoperation.callback.endpoint.template",
	func(in any) (*template.Template, error) {
		s, ok := in.(string)
		if !ok {
			return nil, fmt.Errorf("invalid config type: %T for nexusoperation.callback.endpoint.template, expected string", in)
		}
		if s == "unset" {
			return nil, nil
		}
		return template.New("NexusCallbackURL").Parse(s)
	},
	nil,
	`Controls the template for generating callback URLs included in Nexus operation requests, which are used to deliver
asynchronous completion for external endpoint targets. The template can be used to interpolate the {{.NamepaceName}}
and {{.NamespaceID}} parameters to construct a publicly accessible URL.
Must be set to call external endpoints.`,
)

var ChasmWorkflowOperationsRolloutPercent = dynamicconfig.NewNamespaceIntSetting(
	"nexusoperation.chasmWorkflowOperationsRolloutPercent",
	0,
	`Per-namespace percentage [0,100] of workflows whose new Nexus Operations are created on the CHASM-based
implementation instead of the legacy HSM-based implementation. This setting is only consulted when
enableChasmWorkflowOperations is true and is re-evaluated on every ScheduleNexusOperation command. Membership is
decided by a stable hash of the namespace and workflow ID, so a given workflow consistently lands on the same
implementation across all of its operations and dialing the percentage up is monotonic. Defaults to 0 (conservative
dial-up model): even when enableChasmWorkflowOperations is on, no operations are routed to CHASM until the percentage
is explicitly dialed up.`,
)

var DisallowedOperationHeaders = dynamicconfig.NewGlobalTypedSettingWithConverter(
	"nexusoperation.disallowedHeaders",
	func(in any) ([]string, error) {
		keys, err := dynamicconfig.ConvertStructure[[]string](nil)(in)
		if err != nil {
			return nil, err
		}
		for i, k := range keys {
			keys[i] = strings.ToLower(k)
		}
		return keys, nil
	},
	[]string{
		"request-timeout",
		interceptor.DCRedirectionAPIHeaderName,
		interceptor.DCRedirectionSourceCellHeaderName,
		interceptor.DCRedirectionContextHeaderName,
		headers.CallerNameHeaderName,
		headers.CallerTypeHeaderName,
		headers.CallOriginHeaderName,
		headers.PrincipalTypeHeaderName,
		headers.PrincipalNameHeaderName,
	},
	`Case insensitive list of disallowed header keys for Nexus Operations. ScheduleNexusOperation commands with a
"nexus_header" field that contains any of these disallowed keys will be rejected.`,
)

var EnableChasmWorkflowOperations = dynamicconfig.NewNamespaceBoolSetting(
	"nexusoperation.enableChasmWorkflowOperations",
	false,
	`Feature flag that controls whether the legacy HSM-based implementation (when flag is false; default) or the newer
CHASM-based implementation of Nexus will be used when scheduling new Nexus Operations.`,
)

var Enabled = dynamicconfig.NewNamespaceBoolSetting(
	"nexusoperation.enableStandalone",
	false,
	`Toggles standalone Nexus operation functionality on the server.`,
)

var ErrCancellationAlreadyRequested = serviceerror.NewFailedPrecondition("cancellation already requested")

var ErrOperationAlreadyCompleted = serviceerror.NewFailedPrecondition("operation already completed")

var ErrStandaloneNexusOperationDisabled = serviceerror.NewUnimplemented("Standalone Nexus operation is disabled")

var FrontendModule = fx.Module(
	"chasm.lib.nexusoperation.frontend",
	fx.Provide(configProvider),
	fx.Provide(nexusoperationpb.NewNexusOperationServiceLayeredClient),
	fx.Provide(NewFrontendHandler),
	fx.Provide(newComponentOnlyLibrary),
	fx.Invoke(func(l *componentOnlyLibrary, registry *chasm.Registry) error {

		return registry.Register(l)
	}),
)

var LongPollBuffer = dynamicconfig.NewNamespaceDurationSetting(
	"nexusoperation.longPollBuffer",
	common.DefaultLongPollBuffer,
	`A buffer used to adjust the nexus operation long-poll timeouts.
 Specifically, nexus operation long-poll requests are timed out at a time which leaves at least the buffer's duration
 remaining before the caller's deadline, if permitted by the caller's deadline.`,
)

var LongPollTimeout = dynamicconfig.NewNamespaceDurationSetting(
	"nexusoperation.longPollTimeout",
	common.DefaultLongPollTimeout,
	`Maximum timeout for nexus operation long-poll requests. Actual wait may be shorter to leave
longPollBuffer before the caller deadline.`,
)

var MaxConcurrentOperationsPerWorkflow = dynamicconfig.NewNamespaceIntSetting(
	"nexusoperation.limit.operation.concurrencyPerWorkflow.max",
	2000,
	`Limits the maximum allowed concurrent Nexus Operations for a given workflow execution. Once the limit is reached,
ScheduleNexusOperation commands will be rejected.`,
)

var MaxOperationHeaderSize = dynamicconfig.NewNamespaceIntSetting(
	"nexusoperation.limit.header.size",
	8192,
	`The maximum allowed header size for a Nexus Operation.
ScheduleNexusOperation commands with a "nexus_header" field that exceeds this limit will be rejected.
Uses Go's len() function on header keys and values to determine the total size.`,
)

var MaxOperationNameLength = dynamicconfig.NewNamespaceIntSetting(
	"nexusoperation.limit.operation.name.length",
	1000,
	`Limits the maximum allowed length for a Nexus Operation name. ScheduleNexusOperation commands with an operation name
that exceeds this limit will be rejected.  Uses Go's len() function to determine the length.`,
)

var MaxOperationScheduleToCloseTimeout = dynamicconfig.NewNamespaceDurationSetting(
	"nexusoperation.limit.scheduleToCloseTimeout",
	0,
	`Maximum allowed duration of a Nexus Operation. ScheduleOperation commands that specify no schedule-to-close timeout
or a longer timeout than permitted will have their schedule-to-close timeout capped to this value. 0 implies no limit.`,
)

var MaxOperationTokenLength = dynamicconfig.NewNamespaceIntSetting(
	"nexusoperation.limit.operation.token.length",
	4096,
	`Limits the maximum allowed length for a Nexus Operation token. Tokens returned via start responses or via async
completions that exceed this limit will be rejected. Uses Go's len() function to determine the length.
Leave this limit long enough to fit a workflow ID and namespace name plus padding at minimum since that's what the SDKs
use as the token.`,
)

var MaxReasonLength = dynamicconfig.NewNamespaceIntSetting(
	"nexusoperation.limit.reasonLength",
	1000,
	`Limits the maximum allowed length for a reason string in Nexus operation requests.
Uses Go's len() function to determine the length.`,
)

var MaxServiceNameLength = dynamicconfig.NewNamespaceIntSetting(
	"nexusoperation.limit.service.name.length",
	1000,
	`Limits the maximum allowed length for a Nexus Service name. ScheduleNexusOperation commands with a service name that
exceeds this limit will be rejected.  Uses Go's len() function to determine the length.`,
)

var MetricTagConfiguration = dynamicconfig.NewGlobalTypedSetting(
	"nexusoperation.metrics.tags",
	NexusMetricTagConfig{},
	`Controls which metric tags are included with Nexus operation metrics. This configuration supports:
1. Service name tag - adds the Nexus service name as a metric dimension (IncludeServiceTag)
2. Operation name tag - adds the Nexus operation name as a metric dimension (IncludeOperationTag)
3. Header-based tags - maps values from request headers to metric tags (HeaderTagMappings)

Note: default metric tags (like namespace, endpoint) are always included and not affected by this configuration.
Adding high-cardinality tags (like unique operation names) can significantly increase metric storage requirements and
query complexity. Consider the cardinality impact when enabling these tags.`,
)

var MinDispatchTaskTimeout = dynamicconfig.NewNamespaceDurationSetting(
	"nexusoperation.limit.dispatch.task.timeout.min",
	time.Second,
	`Minimum time remaining for a request to be dispatched to the handler worker. If the remaining request timeout is less
than this value, a timeout error will be returned. Working in conjunction with MinRequestTimeout, both configs help
ensure that the server has enough time to complete a Nexus request.`,
)

var MinRequestTimeout = dynamicconfig.NewNamespaceDurationSetting(
	"nexusoperation.limit.request.timeout.min",
	time.Millisecond*1500,
	`Minimum time remaining for a request to complete for the server to make RPCs. If the remaining request timeout is
less than this value, a non-retryable timeout error will be returned.`,
)

var Module = fx.Module(
	"chasm.lib.nexusoperation",
	fx.Provide(configProvider),
	fx.Provide(commonnexus.NewCallbackTokenGenerator),
	fx.Provide(endpointRegistryProvider),
	fx.Invoke(endpointRegistryLifetimeHooks),
	fx.Provide(defaultNexusTransportProvider),
	fx.Provide(clientProviderFactory),
	fx.Provide(newHandler),
	fx.Provide(newCancellationBackoffTaskHandler),
	fx.Provide(newCancellationInvocationTaskHandler),
	fx.Provide(newOperationBackoffTaskHandler),
	fx.Provide(newOperationInvocationTaskHandler),
	fx.Provide(newOperationScheduleToCloseTimeoutTaskHandler),
	fx.Provide(newOperationScheduleToStartTimeoutTaskHandler),
	fx.Provide(newOperationStartToCloseTimeoutTaskHandler),
	fx.Provide(newLibrary),
	fx.Invoke(register),
)

var NexusOperationCancelCount = metrics.NewCounterDef(
	"nexus_operation_cancel",
	metrics.WithDescription("Nexus Operations cancellations."),
)

var NexusOperationFailedCount = metrics.NewCounterDef(
	"nexus_operation_fail",
	metrics.WithDescription("Nexus Operations failures."),
)

var NexusOperationScheduleToCloseLatency = metrics.NewTimerDef(
	"nexus_operation_schedule_to_close_latency",
	metrics.WithDescription("Duration from Nexus Operation scheduled time to terminal state."),
)

var NexusOperationScheduleToStartLatency = metrics.NewTimerDef(
	"nexus_operation_schedule_to_start_latency",
	metrics.WithDescription("Duration from Nexus Operation scheduled time to started time."),
)

var NexusOperationStartToCloseLatency = metrics.NewTimerDef(
	"nexus_operation_start_to_close_latency",
	metrics.WithDescription("Duration from Nexus Operation started time to completed time. Only emitted for async operations."),
)

var NexusOperationSuccessCount = metrics.NewCounterDef(
	"nexus_operation_success",
	metrics.WithDescription("Nexus Operations successfully completed."),
)

var NexusOperationTerminateCount = metrics.NewCounterDef(
	"nexus_operation_terminate",
	metrics.WithDescription("Nexus Operations that were terminated before completion."),
)

var NexusOperationTimeoutCount = metrics.NewCounterDef(
	"nexus_operation_timeout",
	metrics.WithDescription("Nexus Operations that timed out before completion."),
)

var OperationContextKey = operationContextKeyType{}

var OutboundRequestCounter = metrics.NewCounterDef(
	"nexus_outbound_requests",
	metrics.WithDescription("The number of Nexus outbound requests made by the history service."),
)

var OutboundRequestLatency = metrics.NewTimerDef(
	"nexus_outbound_latency",
	metrics.WithDescription("Latency of outbound Nexus requests made by the history service."),
)

var RequestTimeout = dynamicconfig.NewDestinationDurationSetting(
	"nexusoperation.request.timeout",
	time.Second*10,
	`Timeout for making a single nexus start or cancel request.`,
)

var RetryPolicy = dynamicconfig.NewGlobalTypedSettingWithConverter(
	"nexusoperation.retryPolicy",
	func(in any) (backoff.RetryPolicy, error) {
		cfg, err := dynamicconfig.ConvertStructure(defaultRetryPolicyConfig)(in)
		if err != nil {
			return nil, err
		}
		return cfg.build(), nil
	},
	defaultRetryPolicyConfig.build(),
	`The retry policy for nexus StartOperation or CancelOperation requests for a given operation.`,
)

var TransitionCanceled = chasm.NewTransition(
	[]nexusoperationpb.OperationStatus{
		nexusoperationpb.OPERATION_STATUS_SCHEDULED,
		nexusoperationpb.OPERATION_STATUS_STARTED,
		nexusoperationpb.OPERATION_STATUS_BACKING_OFF,
	},
	nexusoperationpb.OPERATION_STATUS_CANCELED,
	func(o *Operation, ctx chasm.MutableContext, event EventCanceled) error {
		closeTime := ctx.Now(o)
		if event.CompleteTime != nil {
			closeTime = *event.CompleteTime
		}
		o.emitOnCanceledMetrics(ctx, closeTime)

		fromAttempt := o.GetStatus() == nexusoperationpb.OPERATION_STATUS_SCHEDULED
		return o.resolveUnsuccessfully(ctx, event.Failure, closeTime, fromAttempt)
	},
)

var TransitionCancellationFailed = chasm.NewTransition(
	[]nexusoperationpb.CancellationStatus{

		nexusoperationpb.CANCELLATION_STATUS_UNSPECIFIED,
		nexusoperationpb.CANCELLATION_STATUS_SCHEDULED,
	},
	nexusoperationpb.CANCELLATION_STATUS_FAILED,
	func(c *Cancellation, ctx chasm.MutableContext, event EventCancellationFailed) error {
		currentTime := ctx.Now(c)
		c.LastAttemptCompleteTime = timestamppb.New(currentTime)
		c.LastAttemptFailure = event.Failure

		return nil
	},
)

var TransitionCancellationScheduled = chasm.NewTransition(
	[]nexusoperationpb.CancellationStatus{nexusoperationpb.CANCELLATION_STATUS_UNSPECIFIED},
	nexusoperationpb.CANCELLATION_STATUS_SCHEDULED,
	func(c *Cancellation, ctx chasm.MutableContext, event EventCancellationScheduled) error {
		c.Attempt++

		ctx.AddTask(c, chasm.TaskAttributes{
			Destination: event.Destination,
		}, &nexusoperationpb.CancellationTask{
			Attempt: c.Attempt,
		})

		return nil
	},
)

var TransitionCancellationSucceeded = chasm.NewTransition(
	[]nexusoperationpb.CancellationStatus{nexusoperationpb.CANCELLATION_STATUS_SCHEDULED},
	nexusoperationpb.CANCELLATION_STATUS_SUCCEEDED,
	func(c *Cancellation, ctx chasm.MutableContext, event EventCancellationSucceeded) error {
		currentTime := ctx.Now(c)
		c.LastAttemptCompleteTime = timestamppb.New(currentTime)
		c.LastAttemptFailure = nil

		return nil
	},
)

var TransitionFailed = chasm.NewTransition(
	[]nexusoperationpb.OperationStatus{
		nexusoperationpb.OPERATION_STATUS_SCHEDULED,
		nexusoperationpb.OPERATION_STATUS_STARTED,
		nexusoperationpb.OPERATION_STATUS_BACKING_OFF,
	},
	nexusoperationpb.OPERATION_STATUS_FAILED,
	func(o *Operation, ctx chasm.MutableContext, event EventFailed) error {
		closeTime := ctx.Now(o)
		if event.CompleteTime != nil {
			closeTime = *event.CompleteTime
		}

		fromAttempt := o.GetStatus() == nexusoperationpb.OPERATION_STATUS_SCHEDULED
		o.emitOnFailedMetrics(ctx, closeTime)
		return o.resolveUnsuccessfully(ctx, event.Failure, closeTime, fromAttempt)
	},
)

var TransitionScheduled = chasm.NewTransition(
	[]nexusoperationpb.OperationStatus{nexusoperationpb.OPERATION_STATUS_UNSPECIFIED},
	nexusoperationpb.OPERATION_STATUS_SCHEDULED,
	func(o *Operation, ctx chasm.MutableContext, event EventScheduled) error {
		o.Attempt++

		ctx.AddTask(o, chasm.TaskAttributes{Destination: o.GetEndpoint()}, &nexusoperationpb.InvocationTask{
			Attempt: o.Attempt,
		})

		if o.ScheduleToStartTimeout != nil && o.ScheduleToStartTimeout.AsDuration() != 0 {
			deadline := o.ScheduledTime.AsTime().Add(o.ScheduleToStartTimeout.AsDuration())
			ctx.AddTask(o, chasm.TaskAttributes{
				ScheduledTime: deadline,
			}, &nexusoperationpb.ScheduleToStartTimeoutTask{})
		}

		if o.ScheduleToCloseTimeout != nil && o.ScheduleToCloseTimeout.AsDuration() != 0 {
			deadline := o.ScheduledTime.AsTime().Add(o.ScheduleToCloseTimeout.AsDuration())
			ctx.AddTask(o, chasm.TaskAttributes{
				ScheduledTime: deadline,
			}, &nexusoperationpb.ScheduleToCloseTimeoutTask{})
		}

		return nil
	},
)

var TransitionStarted = chasm.NewTransition(
	[]nexusoperationpb.OperationStatus{
		nexusoperationpb.OPERATION_STATUS_SCHEDULED,
		nexusoperationpb.OPERATION_STATUS_BACKING_OFF,
	},
	nexusoperationpb.OPERATION_STATUS_STARTED,
	func(o *Operation, ctx chasm.MutableContext, event EventStarted) error {
		startTime := ctx.Now(o)
		if event.StartTime != nil {
			startTime = *event.StartTime
		}

		o.StartedTime = timestamppb.New(startTime)

		o.LastAttemptCompleteTime = o.StartedTime
		o.LastAttemptFailure = nil

		o.NextAttemptScheduleTime = nil

		o.OperationToken = event.OperationToken

		metricsHandler := o.metricsHandler(ctx)
		NexusOperationScheduleToStartLatency.With(metricsHandler).Record(startTime.Sub(o.GetScheduledTime().AsTime()))

		if o.StartToCloseTimeout != nil && o.StartToCloseTimeout.AsDuration() != 0 {
			deadline := startTime.Add(o.StartToCloseTimeout.AsDuration())
			ctx.AddTask(o, chasm.TaskAttributes{
				ScheduledTime: deadline,
			}, &nexusoperationpb.StartToCloseTimeoutTask{})
		}

		cancellation, ok := o.Cancellation.TryGet(ctx)
		if ok && cancellation.StateMachineState() == nexusoperationpb.CANCELLATION_STATUS_UNSPECIFIED {
			return TransitionCancellationScheduled.Apply(cancellation, ctx, EventCancellationScheduled{
				Destination: o.GetEndpoint(),
			})
		}

		return nil
	},
)

var TransitionSucceeded = chasm.NewTransition(
	[]nexusoperationpb.OperationStatus{
		nexusoperationpb.OPERATION_STATUS_SCHEDULED,
		nexusoperationpb.OPERATION_STATUS_STARTED,
		nexusoperationpb.OPERATION_STATUS_BACKING_OFF,
	},
	nexusoperationpb.OPERATION_STATUS_SUCCEEDED,
	func(o *Operation, ctx chasm.MutableContext, event EventSucceeded) error {

		closeTime := ctx.Now(o)
		if event.CompleteTime != nil {
			closeTime = *event.CompleteTime
		}
		o.NextAttemptScheduleTime = nil
		o.ClosedTime = timestamppb.New(closeTime)

		o.getOrCreateOutcome(ctx).Variant = &nexusoperationpb.OperationOutcome_Successful_{
			Successful: &nexusoperationpb.OperationOutcome_Successful{Result: event.Result},
		}

		o.emitOnSucceededMetrics(ctx, closeTime)

		return nil
	},
)

var TransitionTerminated = chasm.NewTransition(
	[]nexusoperationpb.OperationStatus{
		nexusoperationpb.OPERATION_STATUS_SCHEDULED,
		nexusoperationpb.OPERATION_STATUS_STARTED,
		nexusoperationpb.OPERATION_STATUS_BACKING_OFF,
		nexusoperationpb.OPERATION_STATUS_CANCELED,
	},
	nexusoperationpb.OPERATION_STATUS_TERMINATED,
	func(o *Operation, ctx chasm.MutableContext, event EventTerminated) error {
		closeTime := ctx.Now(o)
		o.TerminateState = &nexusoperationpb.NexusOperationTerminateState{
			RequestId: event.RequestID,
		}
		failure := &failurepb.Failure{
			Message: event.Reason,
			FailureInfo: &failurepb.Failure_TerminatedFailureInfo{
				TerminatedFailureInfo: &failurepb.TerminatedFailureInfo{
					Identity: event.Identity,
				},
			},
		}
		o.emitOnTerminatedMetrics(ctx, closeTime)
		return o.resolveUnsuccessfully(ctx, failure, closeTime, false)
	},
)

var TransitionTimedOut = chasm.NewTransition(
	[]nexusoperationpb.OperationStatus{
		nexusoperationpb.OPERATION_STATUS_SCHEDULED,
		nexusoperationpb.OPERATION_STATUS_STARTED,
		nexusoperationpb.OPERATION_STATUS_BACKING_OFF,
	},
	nexusoperationpb.OPERATION_STATUS_TIMED_OUT,
	func(o *Operation, ctx chasm.MutableContext, event EventTimedOut) error {
		closeTime := ctx.Now(o)
		timeoutType := event.Failure.GetTimeoutFailureInfo().GetTimeoutType().String()
		o.emitOnTimedOutMetrics(ctx, closeTime, timeoutType)
		return o.resolveUnsuccessfully(ctx, event.Failure, closeTime, event.FromAttempt)
	},
)

var UseNewFailureWireFormat = dynamicconfig.NewNamespaceBoolSetting(
	"nexusoperation.useNewFailureWireFormat",
	true,
	`Controls whether to use the new failure wire format via an HTTP header that is attached to StartOperation requests.
Added for safety. Defaults to true. Likely to be removed in future server versions.`,
)