operatorclient

type Client interface {
	Connect(ctx context.Context, req *ConnectRequest) (Session, error)
}

type CloseOpt func(*closeOpts)

type ConnectRequest struct {
	// Name is the operator name, unique per tenant among GRPC operators. The
	// engine upserts the operator row by this name.
	Name string `validate:"required"`

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

	// Labels are worker labels for affinity assignment. Values follow the
	// same conventions as GetActionListenerRequest.Labels (string or int).
	Labels map[string]interface{}

	// ResumeWorker controls whether a reconnect resumes the previous worker
	// id. nil means true.
	ResumeWorker *bool
}

type Opt func(*opts)

type Registration struct {
	TenantId   string
	OperatorId string
	WorkerId   string

	// Resumed reports whether the most recent registration resumed the
	// previous worker. It is false on the first connect.
	Resumed bool
}

type Session interface {
	// Actions starts the receive and heartbeat loops and returns the assigned
	// action stream. The channels close when ctx is cancelled, the session is
	// closed, or the stream fails permanently (reported on the error channel).
	Actions(ctx context.Context) (<-chan *dispatchercontracts.AssignedAction, <-chan error, error)

	// Registration returns the identity assigned by the most recent
	// successful registration.
	Registration() Registration

	// SendStepActionEvent reports task progress. An empty WorkerId is filled
	// from the current registration.
	SendStepActionEvent(ctx context.Context, in *dispatchercontracts.StepActionEvent) (*dispatchercontracts.ActionEventResponse, error)

	// PutWorkflow registers or updates a workflow through the admin service on
	// the same connection and returns the action ids its tasks (including the
	// on-failure task) run, normalized the way the engine stores them. It does
	// not change the worker's action set: pass the ids to AddActions once the
	// operator can run them.
	PutWorkflow(ctx context.Context, wf *v1.CreateWorkflowVersionRequest) (*v1.CreateWorkflowVersionResponse, []string, error)

	// AddActions queues action ids to add to the worker's action set. It
	// never blocks: deltas are coalesced and sent in the background. Call
	// Flush to wait for them to reach the engine.
	AddActions(ids ...string)

	// RemoveActions queues action ids to remove from the worker's action set.
	// It never blocks; see AddActions.
	RemoveActions(ids ...string)

	// Flush waits until every queued delta has been acknowledged by the
	// engine, which acknowledges a delta once it is committed to the worker's
	// action set. A delta whose send failed and has not been retried yet is
	// reported as the send error; it stays queued and is replayed by the next
	// reconnect. The scheduler observes a committed change within about a
	// second.
	Flush(ctx context.Context) error

	// OpenDurableTaskStream opens the OperatorService durable task stream
	// with the session's operator metadata, shaped as the V1Dispatcher
	// stream a durable task listener expects. The register message's worker
	// id is rewritten to the current registration, so a listener built
	// before a reconnect registers the worker the engine now knows. The
	// caller owns the listener it builds over it and stops it before Close.
	OpenDurableTaskStream(ctx context.Context) (v1.V1Dispatcher_DurableTaskClient, error)

	// Pause stops the scheduler assigning to this session's worker. It sends
	// the pause on the Listen stream and returns once the engine has
	// acknowledged it, at which point nothing more is delivered on the stream:
	// an action the scheduler had already assigned goes back to the queue. A
	// caller that drains afterwards therefore holds the last actions it will
	// get. The pause is stream state that the session restores on every
	// reconnect until Resume.
	Pause(ctx context.Context) error

	// Resume lets the scheduler assign to the worker again, acknowledged the
	// same way.
	Resume(ctx context.Context) error

	// Close pauses the worker, waits for the actions already handed to the
	// consumer to be reported, flushes pending deltas with a short timeout,
	// and ends the Listen stream, which deactivates the worker. Pass
	// WithoutDrain to hang up at once instead.
	Close(opts ...CloseOpt) error
}