Skip to content

WorkflowEngine

Defines workflow engine services and an in-memory implementation.

WorkflowEngine registers workflow handlers, runs executions, polls results, resumes suspended runs, executes activities, stores durable deferred results, and schedules durable clocks. WorkflowInstance holds the runtime state for one workflow run. The in-memory layer is useful for tests and local development.

5 exports Added in v4.0.0 Source

Constructors

makeUnsafe

Added in v4.0.0 Source

Builds a typed WorkflowEngine service from a low-level encoded implementation.

When to use

Use when wiring a trusted low-level workflow engine implementation into the typed WorkflowEngine service.

Gotchas

The implementation must correctly persist, resume, and encode workflow state.

Signature

declare function makeUnsafe(options: Encoded): {
  readonly activityExecute: <Success extends Constraint, Error extends Constraint, R>(
    activity: Activity<Success, Error, R>,
    attempt: number,
  ) => Effect<
    Result<Success["Type"], Error["Type"]>,
    never,
    WorkflowInstance | R | Success["DecodingServices"] | Error["DecodingServices"]
  >;
  readonly deferredDone: <Success extends Constraint, Error extends Constraint>(
    deferred: DurableDeferred<Success, Error>,
    options: {
      readonly deferredName: string;
      readonly executionId: string;
      readonly exit: Exit<Success["Type"], Error["Type"]>;
      readonly workflowName: string;
    },
  ) => Effect<void, never, Success["EncodingServices"] | Error["EncodingServices"]>;
  readonly deferredResult: <Success extends Constraint, Error extends Constraint>(
    deferred: DurableDeferred<Success, Error>,
  ) => Effect<Option<Exit<Success["Type"], Error["Type"]>>, never, WorkflowInstance>;
  readonly execute: <
    Name extends string,
    Payload extends AnyStructSchema,
    Success extends Top,
    Error extends Top,
    Discard extends boolean = false,
  >(
    workflow: Workflow<Name, Payload, Success, Error>,
    options: {
      readonly discard?: Discard;
      readonly executionId: string;
      readonly payload: Payload["Type"];
      readonly suspendedRetrySchedule?: Schedule<any, unknown, never, never>;
    },
  ) => Effect<
    Discard extends true ? string : Success["Type"],
    Error["Type"],
    Payload["EncodingServices"] | Success["DecodingServices"] | Error["DecodingServices"]
  >;
  readonly interrupt: (workflow: Any, executionId: string) => Effect<void>;
  readonly interruptUnsafe: (workflow: Any, executionId: string) => Effect<void>;
  readonly poll: <
    Name extends string,
    Payload extends AnyStructSchema,
    Success extends Top,
    Error extends Top,
  >(
    workflow: Workflow<Name, Payload, Success, Error>,
    executionId: string,
  ) => Effect<
    Option<Result<Success["Type"], Error["Type"]>>,
    never,
    Success["DecodingServices"] | Error["DecodingServices"]
  >;
  readonly register: <
    Name extends string,
    Payload extends AnyStructSchema,
    Success extends Top,
    Error extends Top,
    R,
  >(
    workflow: Workflow<Name, Payload, Success, Error>,
    execute: (
      payload: Payload["Type"],
      executionId: string,
    ) => Effect<Success["Type"], Error["Type"], R>,
  ) => Effect<
    void,
    never,
    | Scope
    | Exclude<R, Scope | WorkflowEngine | WorkflowInstance | Execution<Name>>
    | Payload["DecodingServices"]
    | Payload["EncodingServices"]
    | Success["DecodingServices"]
    | Success["EncodingServices"]
    | Error["DecodingServices"]
    | Error["EncodingServices"]
  >;
  readonly resume: (workflow: Any, executionId: string) => Effect<void>;
  readonly scheduleClock: (
    workflow: Any,
    options: {
      readonly clock: DurableClock;
      readonly executionId: string;
    },
  ) => Effect<void>;
};

Layers

layerMemory

Added in v4.0.0 Source

Layer that provides an in-memory WorkflowEngine.

When to use

Use to run tests and local development workflows where durability is not needed.

Gotchas

This layer keeps state only in memory and is not suitable for production workflows that require durability.

Signature

declare const layerMemory: Layer.Layer<WorkflowEngine>;

Services

Encoded interface

Added in v4.0.0 Source

Low-level workflow engine contract that works with encoded payloads and results before makeUnsafe adds typed schema decoding and encoding.

Signature

interface Encoded {
  readonly activityExecute: (
    activity: Any,
    attempt: number,
  ) => Effect<Result<unknown, unknown>, never, WorkflowInstance>;
  readonly deferredDone: (options: {
    readonly deferredName: string;
    readonly executionId: string;
    readonly exit: Exit<unknown, unknown>;
    readonly workflowName: string;
  }) => Effect<void>;
  readonly deferredResult: (
    deferred: Any,
  ) => Effect<Option<Exit<unknown, unknown>>, never, WorkflowInstance>;
  readonly execute: <Discard extends boolean>(
    workflow: Any,
    options: {
      readonly discard: Discard;
      readonly executionId: string;
      readonly parent?: {
        readonly activityState: {
          count: number;
          readonly latch: Latch;
        };
        cause: Cause<never> | undefined;
        readonly executionId: string;
        interrupted: boolean;
        readonly scope: Closeable;
        suspended: boolean;
        readonly workflow: Any;
      };
      readonly payload: object;
    },
  ) => Effect<Discard extends true ? void : Result<unknown, unknown>>;
  readonly interrupt: (workflow: Any, executionId: string) => Effect<void>;
  readonly interruptUnsafe: (workflow: Any, executionId: string) => Effect<void>;
  readonly poll: (workflow: Any, executionId: string) => Effect<Option<Result<unknown, unknown>>>;
  readonly register: (
    workflow: Any,
    execute: (
      payload: object,
      executionId: string,
    ) => Effect<unknown, unknown, WorkflowEngine | WorkflowInstance>,
  ) => Effect<void, never, Scope>;
  readonly resume: (workflow: Any, executionId: string) => Effect<void>;
  readonly scheduleClock: (
    workflow: Any,
    options: {
      readonly clock: DurableClock;
      readonly executionId: string;
    },
  ) => Effect<void>;
}

WorkflowEngine

Added in v4.0.0 Source

Service that represents workflow runtimes, responsible for registering and executing workflows and coordinating activities, durable deferreds, interrupts, resumes, and clocks.

Signature

declare class WorkflowEngine extends Shape<
  "effect/workflow/WorkflowEngine",
  {
    readonly activityExecute: <Success extends Constraint, Error extends Constraint, R>(
      activity: Activity<Success, Error, R>,
      attempt: number,
    ) => Effect<
      Result<Success["Type"], Error["Type"]>,
      never,
      WorkflowInstance | R | Success["DecodingServices"] | Error["DecodingServices"]
    >;
    readonly deferredDone: <Success extends Constraint, Error extends Constraint>(
      deferred: DurableDeferred<Success, Error>,
      options: {
        readonly deferredName: string;
        readonly executionId: string;
        readonly exit: Exit<Success["Type"], Error["Type"]>;
        readonly workflowName: string;
      },
    ) => Effect<void, never, Success["EncodingServices"] | Error["EncodingServices"]>;
    readonly deferredResult: <Success extends Constraint, Error extends Constraint>(
      deferred: DurableDeferred<Success, Error>,
    ) => Effect<Option<Exit<Success["Type"], Error["Type"]>>, never, WorkflowInstance>;
    readonly execute: <
      Name extends string,
      Payload extends AnyStructSchema,
      Success extends Top,
      Error extends Top,
      Discard extends boolean = false,
    >(
      workflow: Workflow<Name, Payload, Success, Error>,
      options: {
        readonly discard?: Discard;
        readonly executionId: string;
        readonly payload: Payload["Type"];
        readonly suspendedRetrySchedule?: Schedule<any, unknown, never, never>;
      },
    ) => Effect<
      Discard extends true ? string : Success["Type"],
      Error["Type"],
      Payload["EncodingServices"] | Success["DecodingServices"] | Error["DecodingServices"]
    >;
    readonly interrupt: (workflow: Any, executionId: string) => Effect<void>;
    readonly interruptUnsafe: (workflow: Any, executionId: string) => Effect<void>;
    readonly poll: <
      Name extends string,
      Payload extends AnyStructSchema,
      Success extends Top,
      Error extends Top,
    >(
      workflow: Workflow<Name, Payload, Success, Error>,
      executionId: string,
    ) => Effect<
      Option<Result<Success["Type"], Error["Type"]>>,
      never,
      Success["DecodingServices"] | Error["DecodingServices"]
    >;
    readonly register: <
      Name extends string,
      Payload extends AnyStructSchema,
      Success extends Top,
      Error extends Top,
      R,
    >(
      workflow: Workflow<Name, Payload, Success, Error>,
      execute: (
        payload: Payload["Type"],
        executionId: string,
      ) => Effect<Success["Type"], Error["Type"], R>,
    ) => Effect<
      void,
      never,
      | Scope
      | Exclude<R, Scope | WorkflowEngine | WorkflowInstance | Execution<Name>>
      | Payload["DecodingServices"]
      | Payload["EncodingServices"]
      | Success["DecodingServices"]
      | Success["EncodingServices"]
      | Error["DecodingServices"]
      | Error["EncodingServices"]
    >;
    readonly resume: (workflow: Any, executionId: string) => Effect<void>;
    readonly scheduleClock: (
      workflow: Any,
      options: {
        readonly clock: DurableClock;
        readonly executionId: string;
      },
    ) => Effect<void>;
  },
  this
> {
  constructor(_: never);
}

WorkflowInstance

Added in v4.0.0 Source

Service that contains workflow runtime state for one execution.

When to use

Use to read or update workflow execution, suspension, interruption, lifetime, failure, and activity coordination state inside workflow engine internals.

Details

The service stores the execution ID, workflow definition, long-lived scope, suspension and interruption flags, the stored failure cause, and activity coordination state for a single workflow run.

Signature

declare class WorkflowInstance extends Shape<
  "effect/workflow/WorkflowEngine/WorkflowInstance",
  {
    readonly activityState: {
      count: number;
      readonly latch: Latch;
    };
    cause: Cause<never> | undefined;
    readonly executionId: string;
    interrupted: boolean;
    readonly scope: Closeable;
    suspended: boolean;
    readonly workflow: Any;
  },
  this
> {
  constructor(_: never);
  static initial(
    workflow: Any,
    executionId: string,
    scope: Closeable,
  ): {
    readonly activityState: {
      count: number;
      readonly latch: Latch;
    };
    cause: Cause<never> | undefined;
    readonly executionId: string;
    interrupted: boolean;
    readonly scope: Closeable;
    suspended: boolean;
    readonly workflow: Any;
  };
}