operator

type ActionHandler interface {
	HandleAction(ctx context.Context, action *contracts.AssignedAction) error
}

type DAGStepTriggerRequest struct {
	ParentTaskExternalId uuid.UUID
	InvocationCount      int32
	WorkflowName         string
	// WorkflowVersionId pins triggering to the DAG's original version.
	WorkflowVersionId   uuid.UUID
	ActionId            string
	ChildIndex          int32
	Input               string
	AdditionalMetadata  []byte
	DagParentTaskRunIds []uuid.UUID
	IsSkipped           bool
	IsCancelled         bool
	DesiredWorkerLabels []*sqlcv1.GetDesiredLabelsRow

	// ParentReExecuted forces the step to re-run during a replay when any of its parents
	// re-executed this invocation.
	ParentReExecuted bool
}

type DAGStepTriggerResult struct {
	NodeId                int64
	BranchId              int64
	WorkflowRunExternalId uuid.UUID

	IsSatisfied   bool
	ResultPayload []byte
	IsFailure     bool
	ErrorMessage  *string

	// ReExecuted is true when the step actually runs this invocation rather than being
	// satisfied from the log.
	ReExecuted bool
}

type DurableChannel interface {
	Send(ctx context.Context, req *v1.DurableTaskRequest) error
	Recv(ctx context.Context) (*v1.DurableTaskResponse, error)

	// ExpectEntry registers an entry of the invocation whose completion the operator awaits
	// and which no request on this channel acknowledged. A completion that already arrived is
	// delivered on the next Recv; one that arrives later is delivered as it comes. Registering
	// an entry twice, or after its completion was delivered, changes nothing.
	ExpectEntry(branchId, nodeId int64) error

	Close() error
}

type Host interface {
	Open(ctx context.Context, id Identity, opts OpenOpts) (Session, error)
}

type Identity struct {
	TenantId uuid.UUID

	// OperatorId is an existing operator row. The host takes the name, kind and leasing manager from
	// the row and points the row's worker_id at the session's worker, so the claimer keeps
	// recognising the assignment.
	OperatorId *uuid.UUID

	// Name, Kind and LeasingManager identify a row the host upserts. Kind defaults to GRPC, the only
	// kind a host registers by name, and LeasingManager to SELF, the only leasing manager the gRPC host can
	// register: a row it opened is kept alive by its stream, not by the claimer.
	Name           string
	Kind           sqlcv1.V1OperatorKind
	LeasingManager sqlcv1.V1OperatorLeasingManager
}

type OpenOpts struct {
	// Handler receives the actions assigned to the session's worker. Required.
	Handler ActionHandler

	// Actions is the initial action set, linked before Open returns so the worker is never
	// live with an action set the caller did not ask for. Later changes go through
	// Session.AddActions and RemoveActions.
	Actions []string

	// SlotConfig maps slot type to max units. The engine defaults it to {"default": 100}.
	SlotConfig map[string]int32

	// Labels are worker labels for affinity assignment (string or int values).
	Labels map[string]interface{}

	// RuntimeInfo describes the process for the dashboard. The gRPC host reports its own
	// process and ignores it.
	RuntimeInfo *contracts.RuntimeInfo

	// ResumeWorkerId names a previous worker of the same operator to resume instead of
	// creating one. Not every host supports it (ErrNotSupported).
	ResumeWorkerId *uuid.UUID

	// WorkerName names the worker row; it defaults to the operator name. Not every host
	// supports it (ErrNotSupported).
	WorkerName string
}

type Operator interface {
	ActionHandler

	Start(ctx context.Context, s Session) error
	Drain(ctx context.Context)
}

type Registration struct {
	TenantId   uuid.UUID
	OperatorId uuid.UUID
	WorkerId   uuid.UUID

	// Resumed reports whether WorkerId is the worker ResumeWorkerId named.
	Resumed bool
}

type Session interface {
	Registration() Registration

	// AddActions adds ids to the worker's action set; ids already in the set are ignored. A
	// host may apply the delta asynchronously; Flush waits for it.
	AddActions(ctx context.Context, ids []string) error

	// RemoveActions removes ids from the worker's action set; ids not in the set are ignored.
	RemoveActions(ctx context.Context, ids []string) error

	// Flush returns once every delta issued so far is committed by the engine and reports the
	// failure, if any. In-process deltas are synchronous and Flush returns at once.
	Flush(ctx context.Context) error

	// PutWorkflow registers or updates a workflow for the session's tenant and returns the
	// action ids its tasks derive, normalized the way the engine stores them. It does not
	// change the worker's action set.
	PutWorkflow(ctx context.Context, wf *v1.CreateWorkflowVersionRequest) ([]string, error)

	// SendStepActionEvent reports task progress (STARTED, COMPLETED, FAILED, CANCELLED). An
	// empty WorkerId is filled from the registration; another worker's id is refused.
	SendStepActionEvent(ctx context.Context, ev *contracts.StepActionEvent) error

	// OpenDurable opens one invocation's request and response pipe. The register_worker
	// handshake is done by the host; responses that arrive before it completes are held,
	// bounded, and delivered after the ack that names their entry. ctx bounds the handshake;
	// the invocation itself ends on the channel's Close.
	OpenDurable(ctx context.Context, taskExternalId uuid.UUID, invocation int32) (DurableChannel, error)

	// Pause stops the scheduler assigning to the worker and returns once the pause is
	// committed, so a caller that drains afterwards knows no further work will arrive.
	Pause(ctx context.Context) error

	// Close ends the session: the worker is paused if it is not already, the session is
	// released so nothing further is delivered, and the worker is deactivated, fenced on the
	// session id. A host's teardown order is Pause, the operator's Drain, then Close.
	Close(ctx context.Context) error

	// Done is closed once the session no longer serves its worker: after Close, or when the
	// host gave up on it. A host recovers from transient failures on its own (the gRPC host
	// reconnects, re-reads the tenant's token and restores the action set); it gives up on a
	// failure no retry can fix, such as a token source that no longer has a token for the
	// tenant. The caller that owns the session then closes it and opens another.
	Done() <-chan struct{}

	// Err reports why Done closed: nil after Close, the terminal failure when the host gave
	// up. It is nil while the session is live.
	Err() error
}

type SharedOperator[T any] struct {
	// contains filtered or unexported fields
}

type TaskEventWriter interface {
	// CancelTaskWithReason reports a cancelled task with a custom cancellation reason. It is
	// the engine-internal writer behind SendCancelledWithMessage, distinct from the CANCELLED
	// step action event every host offers through Session.SendStepActionEvent, and it is not
	// on the gRPC surface: an out-of-process operator has no equivalent.
	CancelTaskWithReason(ctx context.Context, tenantId uuid.UUID, request *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)

	TriggerDAGStep(ctx context.Context, tenantId uuid.UUID, req *DAGStepTriggerRequest) (*DAGStepTriggerResult, error)

	// CancelDAGChildren cancels already-triggered children when the orchestrator is cancelled.
	CancelDAGChildren(ctx context.Context, tenantId uuid.UUID, taskExternalIds []uuid.UUID) error
}