.opencode/node_modules/effect/dist/unstable/cluster/ClusterWorkflowEngine.js

/**
 * The cluster workflow engine runs durable workflows on top of cluster sharding
 * and message storage. It adapts `WorkflowEngine.WorkflowEngine` so workflow
 * executions, activities, deferred completions, resumes, interrupts, and durable
 * clock wakeups are represented as persisted cluster entity messages.
 *
 * @since 4.0.0
 */
import * as Context from "../../Context.js";
import * as DateTime from "../../DateTime.js";
import * as Duration from "../../Duration.js";
import * as Effect from "../../Effect.js";
import * as Exit from "../../Exit.js";
import * as Fiber from "../../Fiber.js";
import * as Latch from "../../Latch.js";
import * as Layer from "../../Layer.js";
import * as Option from "../../Option.js";
import * as PrimaryKey from "../../PrimaryKey.js";
import * as RcMap from "../../RcMap.js";
import * as Schedule from "../../Schedule.js";
import * as Schema from "../../Schema.js";
import * as Headers from "../http/Headers.js";
import * as Rpc from "../rpc/Rpc.js";
import { ClientAbort } from "../rpc/RpcSchema.js";
import * as Activity from "../workflow/Activity.js";
import * as DurableClock from "../workflow/DurableClock.js";
import * as DurableDeferred from "../workflow/DurableDeferred.js";
import * as Workflow from "../workflow/Workflow.js";
import * as WorkflowEngine from "../workflow/WorkflowEngine.js";
import * as ClusterSchema from "./ClusterSchema.js";
import * as DeliverAt from "./DeliverAt.js";
import * as Entity from "./Entity.js";
import * as EntityAddress from "./EntityAddress.js";
import * as EntityId from "./EntityId.js";
import * as EntityType from "./EntityType.js";
import * as Envelope from "./Envelope.js";
import * as Message from "./Message.js";
import { MessageStorage } from "./MessageStorage.js";
import * as Reply from "./Reply.js";
import * as Sharding from "./Sharding.js";
import * as Snowflake from "./Snowflake.js";
/**
 * Creates a `WorkflowEngine` implementation backed by cluster sharding and
 * message storage.
 *
 * **Details**
 *
 * Workflow executions, activities, deferred completions, resumes, interrupts,
 * and durable clock wakeups are coordinated through persisted cluster entities.
 *
 * @category constructors
 * @since 4.0.0
 */
export const make = /*#__PURE__*/Effect.gen(function* () {
  const sharding = yield* Sharding.Sharding;
  const storage = yield* MessageStorage;
  const workflows = new Map();
  const entities = new Map();
  const partialEntities = new Map();
  const ensureEntity = workflow => {
    let entity = entities.get(workflow._tag);
    if (!entity) {
      entity = makeWorkflowEntity(workflow);
      workflows.set(workflow._tag, workflow);
      entities.set(workflow._tag, entity);
    }
    return entity;
  };
  const ensurePartialEntity = workflowName => {
    let entity = partialEntities.get(workflowName);
    if (!entity) {
      entity = makePartialWorkflowEntity(workflowName);
      partialEntities.set(workflowName, entity);
    }
    return entity;
  };
  const activities = new Map();
  const interruptedActivities = new Set();
  const activityLatches = new Map();
  const clients = yield* RcMap.make({
    lookup: Effect.fnUntraced(function* (workflowName) {
      const entity = entities.get(workflowName);
      if (!entity) {
        return yield* Effect.die(`Workflow ${workflowName} not registered`);
      }
      return yield* entity.client;
    }),
    idleTimeToLive: "5 minutes"
  });
  const clientsPartial = yield* RcMap.make({
    lookup: Effect.fnUntraced(function* (workflowName) {
      const entity = entities.get(workflowName) ?? ensurePartialEntity(workflowName);
      return yield* entity.client;
    }),
    idleTimeToLive: "5 minutes"
  });
  const entityAddressFor = options => {
    const shardGroup = Context.get(options.workflow.annotations, ClusterSchema.ShardGroup)(options.executionId);
    const entityId = EntityId.make(options.executionId);
    return EntityAddress.make({
      entityType: EntityType.make(options.entityType),
      entityId,
      shardId: sharding.getShardId(entityId, shardGroup)
    });
  };
  const sendDiscard = Effect.fnUntraced(function* (options) {
    const payload = options.rpc.payloadSchema.make(options.payload);
    const envelope = Envelope.makeRequest({
      requestId: yield* sharding.getSnowflake,
      address: options.address,
      tag: options.rpc._tag,
      payload,
      headers: Headers.empty
    });
    yield* sharding.sendOutgoing(new Message.OutgoingRequest({
      envelope,
      context: Context.empty(),
      lastReceivedReply: Option.none(),
      rpc: options.rpc,
      respond: () => Effect.void,
      annotations: Context.get(options.rpc.annotations, ClusterSchema.Dynamic)(options.rpc.annotations, envelope)
    }), true);
  });
  const requestIdFor = Effect.fnUntraced(function* (options) {
    const address = entityAddressFor(options);
    return yield* storage.requestIdForPrimaryKey({
      address,
      tag: options.tag,
      id: options.id
    });
  });
  const replyForRequestId = Effect.fnUntraced(function* (requestId) {
    const replies = yield* storage.repliesForUnfiltered([requestId]);
    const last = replies[replies.length - 1];
    if (last && last._tag === "WithExit") {
      return Option.some(last);
    }
    return Option.none();
  });
  const requestReply = Effect.fnUntraced(function* (options) {
    const requestId = yield* requestIdFor(options);
    if (Option.isNone(requestId)) {
      return Option.none();
    }
    return yield* replyForRequestId(requestId.value);
  });
  const resetActivityAttempt = Effect.fnUntraced(function* (options) {
    const requestId = yield* requestIdFor({
      workflow: options.workflow,
      entityType: `Workflow/${options.workflow._tag}`,
      executionId: options.executionId,
      tag: "activity",
      id: activityPrimaryKey(options.activity.name, options.attempt)
    });
    if (Option.isNone(requestId)) return;
    yield* sharding.reset(requestId.value);
  }, Effect.retry({
    times: 3,
    schedule: Schedule.exponential(250)
  }), Effect.orDie);
  const clearClock = Effect.fnUntraced(function* (options) {
    const shardGroup = Context.get(options.workflow.annotations, ClusterSchema.ShardGroup)(options.executionId);
    const entityId = EntityId.make(options.executionId);
    const shardId = sharding.getShardId(entityId, shardGroup);
    const clockAddress = EntityAddress.make({
      entityType: ClockEntity.type,
      entityId,
      shardId
    });
    yield* storage.clearAddress(clockAddress);
  });
  const resume = Effect.fnUntraced(function* (workflow, executionId) {
    const maybeReply = yield* requestReply({
      workflow,
      entityType: `Workflow/${workflow._tag}`,
      executionId,
      tag: "run",
      id: ""
    });
    const maybeSuspended = Option.filter(maybeReply, reply => reply.exit._tag === "Success" && reply.exit.value._tag === "Suspended");
    if (Option.isNone(maybeSuspended)) return;
    yield* sharding.reset(Snowflake.Snowflake(maybeSuspended.value.requestId));
    yield* sharding.pollStorage;
  });
  const sendResumeParent = Effect.fnUntraced(function* (options) {
    const requestId = yield* requestIdFor({
      workflow: workflows.get(options.workflowName),
      entityType: `Workflow/${options.workflowName}`,
      executionId: options.executionId,
      tag: "resume",
      id: ""
    });
    if (Option.isNone(requestId)) {
      const client = (yield* RcMap.get(clientsPartial, options.workflowName))(options.executionId);
      return yield* client.resume({}, {
        discard: true
      });
    }
    const reply = yield* replyForRequestId(requestId.value);
    if (Option.isNone(reply)) return;
    yield* sharding.reset(requestId.value);
  }, Effect.scoped);
  const interrupt = Effect.fnUntraced(function* (workflow, executionId) {
    ensureEntity(workflow);
    const requestId = yield* requestIdFor({
      workflow,
      entityType: `Workflow/${workflow._tag}`,
      executionId,
      tag: "run",
      id: ""
    });
    if (Option.isNone(requestId)) {
      return Option.none();
    }
    const reply = yield* replyForRequestId(requestId.value);
    const nonSuspendedReply = Option.filter(reply, reply => reply.exit._tag !== "Success" || reply.exit.value._tag !== "Suspended");
    if (Option.isSome(nonSuspendedReply)) {
      return Option.none();
    }
    yield* engine.deferredDone(InterruptSignal, {
      workflowName: workflow._tag,
      executionId,
      deferredName: InterruptSignal.name,
      exit: Exit.void
    });
    return requestId;
  }, Effect.retry({
    while: e => e._tag === "PersistenceError",
    times: 3,
    schedule: Schedule.exponential(250)
  }), Effect.orDie);
  const engine = WorkflowEngine.makeUnsafe({
    register: (workflow, execute) => Effect.suspend(() => sharding.registerEntity(ensureEntity(workflow), Effect.gen(function* () {
      const address = yield* Entity.CurrentAddress;
      const executionId = address.entityId;
      return {
        run: request => {
          const instance = WorkflowEngine.WorkflowInstance.initial(workflow, executionId);
          const payload = request.payload;
          let parent;
          if (payload[payloadParentKey]) {
            parent = payload[payloadParentKey];
          }
          return execute(workflow.payloadSchema.make(payload), executionId).pipe(Effect.onExit(exit => {
            const suspendOnFailure = Context.get(workflow.annotations, Workflow.SuspendOnFailure);
            if (!instance.suspended && !(suspendOnFailure && exit._tag === "Failure")) {
              return parent ? ensureSuccess(sendResumeParent(parent)) : Effect.void;
            }
            return engine.deferredResult(InterruptSignal).pipe(Effect.flatMap(maybeExit => {
              if (Option.isNone(maybeExit)) {
                return Effect.void;
              }
              instance.suspended = false;
              instance.interrupted = true;
              return Effect.andThen(Effect.ignore(clearClock({
                workflow,
                executionId
              })), Effect.withFiber(fiber => Effect.interruptible(Fiber.interrupt(fiber))));
            }), Effect.orDie);
          }), Workflow.intoResult, Effect.provideService(WorkflowEngine.WorkflowInstance, instance));
        },
        activity(request) {
          const payload = request.payload;
          const activityId = `${executionId}/${payload.name}`;
          const instance = WorkflowEngine.WorkflowInstance.initial(workflow, executionId);
          interruptedActivities.delete(activityId);
          return Effect.gen(function* () {
            let entry = activities.get(activityId);
            while (!entry) {
              const latch = Latch.makeUnsafe();
              activityLatches.set(activityId, latch);
              yield* latch.await;
              entry = activities.get(activityId);
            }
            const contextMap = new Map(entry.context.mapUnsafe);
            contextMap.set(Activity.CurrentAttempt.key, payload.attempt);
            contextMap.set(WorkflowEngine.WorkflowInstance.key, instance);
            return yield* entry.activity.executeEncoded.pipe(Effect.provideContext(Context.makeUnsafe(contextMap)));
          }).pipe(Workflow.intoResult, Effect.catchCause(cause => {
            // we only want to store interrupts as suspends when the
            // client requested it
            const suspend = cause.reasons.some(f => f._tag === "Interrupt" && f.annotations.has(ClientAbort.key));
            if (suspend) {
              interruptedActivities.add(activityId);
              return Effect.succeed(new Workflow.Suspended({}));
            }
            return Effect.failCause(cause);
          }), Effect.provideService(WorkflowEngine.WorkflowInstance, instance), Effect.provideService(Activity.CurrentAttempt, payload.attempt), Effect.ensuring(Effect.sync(() => {
            activities.delete(activityId);
          })), Rpc.wrap({
            fork: true,
            uninterruptible: true
          }));
        },
        deferred: Effect.fnUntraced(function* (request) {
          const payload = request.payload;
          yield* ensureSuccess(resume(workflow, executionId));
          return payload.exit;
        }),
        resume: () => ensureSuccess(resume(workflow, executionId))
      };
    }))),
    execute: (workflow, {
      discard,
      executionId,
      parent,
      payload
    }) => {
      ensureEntity(workflow);
      return RcMap.get(clients, workflow._tag).pipe(Effect.flatMap(make => make(executionId).run(parent ? {
        ...payload,
        [payloadParentKey]: {
          workflowName: parent.workflow._tag,
          executionId: parent.executionId
        }
      } : payload, {
        discard
      })), Effect.orDie, Effect.scoped);
    },
    poll: Effect.fnUntraced(function* (workflow, executionId) {
      const entity = ensureEntity(workflow);
      const exitSchema = Schema.toCodecJson(Rpc.exitSchema(entity.protocol.requests.get("run")));
      const reply = yield* requestReply({
        workflow,
        entityType: `Workflow/${workflow._tag}`,
        executionId,
        tag: "run",
        id: ""
      });
      if (Option.isNone(reply)) return Option.none();
      const exit = yield* Schema.decodeUnknownEffect(exitSchema)(reply.value.exit);
      return Option.some(yield* exit);
    }, Effect.orDie),
    interrupt: (workflow, executionId) => Effect.asVoid(interrupt(workflow, executionId)),
    interruptUnsafe: Effect.fnUntraced(function* (workflow, executionId) {
      const requestId = yield* interrupt(workflow, executionId);
      if (Option.isNone(requestId)) return;
      const entity = ensureEntity(workflow);
      const runRpc = entity.protocol.requests.get("run");
      yield* Effect.orDie(sharding.sendOutgoing(new Message.OutgoingEnvelope({
        rpc: runRpc,
        envelope: new Envelope.Interrupt({
          id: yield* sharding.getSnowflake,
          address: entityAddressFor({
            workflow,
            entityType: `Workflow/${workflow._tag}`,
            executionId
          }),
          requestId: requestId.value
        })
      }), false));
    }),
    resume: (workflow, executionId) => ensureSuccess(resume(workflow, executionId)),
    activityExecute: Effect.fnUntraced(function* (activity, attempt) {
      const services = yield* Effect.context();
      const instance = Context.get(services, WorkflowEngine.WorkflowInstance);
      yield* Effect.annotateCurrentSpan("executionId", instance.executionId);
      const activityId = `${instance.executionId}/${activity.name}`;
      const client = (yield* RcMap.get(clientsPartial, instance.workflow._tag))(instance.executionId);
      while (true) {
        if (!activities.has(activityId)) {
          activities.set(activityId, {
            activity,
            context: services
          });
          const latch = activityLatches.get(activityId);
          if (latch) {
            yield* latch.open;
            activityLatches.delete(activityId);
          }
        }
        const result = yield* Effect.orDie(client.activity({
          name: activity.name,
          attempt,
          withTransaction: Context.get(activity.annotations, ClusterSchema.WithTransaction)
        }));
        // If the activity has suspended and did not execute, we need to resume
        // it by resetting the attempt and re-executing.
        if (result._tag === "Suspended" && (activities.has(activityId) || interruptedActivities.has(activityId))) {
          yield* resetActivityAttempt({
            workflow: instance.workflow,
            executionId: instance.executionId,
            activity,
            attempt
          });
          continue;
        }
        activities.delete(activityId);
        return result;
      }
    }, Effect.scoped),
    deferredResult: deferred => WorkflowEngine.WorkflowInstance.pipe(Effect.flatMap(instance => requestReply({
      workflow: instance.workflow,
      entityType: `Workflow/${instance.workflow._tag}`,
      executionId: instance.executionId,
      tag: "deferred",
      id: deferred.name
    })), Effect.map(reply => {
      if (Option.isNone(reply)) {
        return Option.none();
      }
      const decoded = decodeDeferredWithExit(reply.value);
      return Option.some(decoded.exit._tag === "Success" ? decoded.exit.value : decoded.exit);
    }), Effect.retry({
      while: e => e._tag === "PersistenceError",
      times: 3,
      schedule: Schedule.exponential(250)
    }), Effect.orDie),
    deferredDone: Effect.fnUntraced(function* ({
      deferredName,
      executionId,
      exit,
      workflowName
    }) {
      const workflow = workflows.get(workflowName);
      if (workflow) {
        return yield* Effect.orDie(sendDiscard({
          rpc: DeferredRpc,
          address: entityAddressFor({
            workflow,
            entityType: `Workflow/${workflowName}`,
            executionId
          }),
          payload: {
            name: deferredName,
            exit
          }
        }));
      }
      const client = yield* RcMap.get(clientsPartial, workflowName);
      return yield* Effect.orDie(client(executionId).deferred({
        name: deferredName,
        exit
      }, {
        discard: true
      }));
    }, Effect.scoped),
    scheduleClock(workflow, options) {
      return DateTime.now.pipe(Effect.flatMap(now => sendDiscard({
        rpc: ClockRpc,
        address: entityAddressFor({
          workflow,
          entityType: ClockEntity.type,
          executionId: options.executionId
        }),
        payload: {
          name: options.clock.name,
          workflowName: workflow._tag,
          wakeUp: DateTime.addDuration(now, options.clock.duration)
        }
      })), Effect.orDie);
    }
  });
  return engine;
});
const retryPolicy = /*#__PURE__*/Schedule.exponential(200, 1.5).pipe(/*#__PURE__*/Schedule.either(/*#__PURE__*/Schedule.spaced("1 minute")));
const ensureSuccess = effect => effect.pipe(Effect.sandbox, Effect.retry(retryPolicy), Effect.orDie);
const AnyOrVoid = /*#__PURE__*/Schema.Union([Schema.Undefined, Schema.Any]);
const ExitUnknown = /*#__PURE__*/Schema.Exit(AnyOrVoid, AnyOrVoid, Schema.Any);
const ActivityRpc = /*#__PURE__*/Rpc.make("activity", {
  payload: {
    name: Schema.String,
    attempt: Schema.Number,
    withTransaction: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false)))
  },
  primaryKey: ({
    attempt,
    name
  }) => activityPrimaryKey(name, attempt),
  success: Workflow.Result({
    success: AnyOrVoid,
    error: AnyOrVoid
  })
}).annotate(ClusterSchema.Persisted, true).annotate(ClusterSchema.Dynamic, (annotations, request) => request.payload.withTransaction ? Context.add(annotations, ClusterSchema.WithTransaction, true) : annotations);
const DeferredRpc = /*#__PURE__*/Rpc.make("deferred", {
  payload: {
    name: Schema.String,
    exit: ExitUnknown
  },
  primaryKey: ({
    name
  }) => name,
  success: ExitUnknown
}).annotate(ClusterSchema.Persisted, true).annotate(ClusterSchema.Uninterruptible, true);
const decodeDeferredWithExit = /*#__PURE__*/Schema.decodeSync(/*#__PURE__*/Schema.toCodecJson(/*#__PURE__*/Reply.WithExit.schema(DeferredRpc)));
const ResumeRpc = /*#__PURE__*/Rpc.make("resume", {
  payload: {},
  primaryKey: () => ""
}).annotate(ClusterSchema.Persisted, true).annotate(ClusterSchema.Uninterruptible, "server");
const payloadParentKey = "~effect/cluster/ClusterWorkflowEngine/payloadParentKey";
const makeWorkflowEntity = workflow => Entity.make(`Workflow/${workflow._tag}`, [Rpc.make("run", {
  payload: {
    ...workflow.payloadSchema.fields,
    [payloadParentKey]: Schema.optional(Schema.Struct({
      workflowName: Schema.String,
      executionId: Schema.String
    }))
  },
  primaryKey: () => "",
  success: Workflow.Result({
    success: workflow.successSchema,
    error: workflow.errorSchema
  })
}).annotate(ClusterSchema.Persisted, true).annotate(ClusterSchema.Uninterruptible, true), DeferredRpc, ResumeRpc, ActivityRpc]).annotateMerge(workflow.annotations);
const makePartialWorkflowEntity = workflowName => Entity.make(`Workflow/${workflowName}`, [DeferredRpc, ResumeRpc, ActivityRpc]);
const activityPrimaryKey = (activity, attempt) => `${activity}/${attempt}`;
class ClockPayload extends /*#__PURE__*/Schema.Class(`Workflow/DurableClock/Run`)({
  name: Schema.String,
  workflowName: Schema.String,
  wakeUp: Schema.DateTimeUtcFromMillis
}) {
  [PrimaryKey.symbol]() {
    return this.name;
  }
  [DeliverAt.symbol]() {
    return this.wakeUp;
  }
}
const ClockRpc = /*#__PURE__*/Rpc.make("run", {
  payload: ClockPayload
}).annotate(ClusterSchema.Persisted, true).annotate(ClusterSchema.Uninterruptible, true);
const ClockEntity = /*#__PURE__*/Entity.make("Workflow/-/DurableClock", [ClockRpc]);
const ClockEntityLayer = /*#__PURE__*/ClockEntity.toLayer(/*#__PURE__*/Effect.gen(function* () {
  const engine = yield* WorkflowEngine.WorkflowEngine;
  const address = yield* Entity.CurrentAddress;
  const executionId = address.entityId;
  return {
    run(request) {
      const deferred = DurableClock.make({
        name: request.payload.name,
        duration: Duration.zero
      }).deferred;
      return ensureSuccess(engine.deferredDone(deferred, {
        workflowName: request.payload.workflowName,
        executionId,
        deferredName: deferred.name,
        exit: Exit.void
      }));
    }
  };
}));
const InterruptSignal = /*#__PURE__*/DurableDeferred.make("Workflow/InterruptSignal");
/**
 * Layer that provides `WorkflowEngine.WorkflowEngine` using the cluster workflow
 * engine implementation.
 *
 * **Details**
 *
 * It requires cluster sharding and message storage, and also registers the
 * durable clock entity used for workflow wakeups.
 *
 * @category layers
 * @since 4.0.0
 */
export const layer = /*#__PURE__*/ClockEntityLayer.pipe(/*#__PURE__*/Layer.provideMerge(/*#__PURE__*/Layer.effect(WorkflowEngine.WorkflowEngine)(make)));
//# sourceMappingURL=ClusterWorkflowEngine.js.map