1/**
2 * Defines named effects whose results can be stored by a workflow engine.
3 *
4 * An `Activity` is an `Effect` with a stable name and schemas for its success
5 * and error values. `make` wraps an effect so the `WorkflowEngine` can execute
6 * it, store its result, or replay that result during a workflow run. This module
7 * also includes helpers for retry attempts, idempotency keys, and durable races.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import type { NonEmptyReadonlyArray } from "../Array.ts"
13import * as Cause from "../Cause.ts"
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import * as Effectable from "../Effectable.ts"
17import { dual } from "../Function.ts"
18import * as Schedule from "../Schedule.ts"
19import * as Schema from "../Schema.ts"
20import type { Scope } from "../Scope.ts"
21import type * as Types from "../Types.ts"
22import * as DurableDeferred from "./DurableDeferred.ts"
23import { makeHashDigest } from "./internal/crypto.ts"
24import * as Workflow from "./Workflow.ts"
25import type { WorkflowEngine, WorkflowInstance } from "./WorkflowEngine.ts"
26
27const TypeId = "~effect/workflow/Activity"
28
29/**
30 * Durable workflow activity that behaves as an `Effect` and records its name,
31 * result schemas, annotations, and encoded execution form for the workflow
32 * engine.
33 *
34 * @stability unstable
35 * @category models
36 * @since 4.0.0
37 */
38export interface Activity<
39 Success extends Schema.Constraint = Schema.Void,
40 Error extends Schema.Constraint = Schema.Never,
41 R = never
42> extends
43 Effect.Effect<
44 Success["Type"],
45 Error["Type"],
46 Success["DecodingServices"] | Error["DecodingServices"] | R | WorkflowEngine | WorkflowInstance
47 >
48{
49 readonly [TypeId]: typeof TypeId
50 readonly name: string
51 readonly successSchema: Success
52 readonly errorSchema: Error
53 readonly exitSchema: Schema.Exit<Success, Error, Schema.Defect>
54 readonly exitSchemaPartial: Schema.Exit<Success, Error, Schema.Unknown>
55 readonly annotations: Context.Context<never>
56 annotate<I, S>(
57 key: Context.Key<I, S>,
58 value: S
59 ): Activity<Success, Error, R>
60 annotateMerge<I>(
61 annotations: Context.Context<I>
62 ): Activity<Success, Error, R>
63 readonly execute: Effect.Effect<
64 Success["Type"],
65 Error["Type"],
66 | Success["DecodingServices"]
67 | Success["EncodingServices"]
68 | Error["DecodingServices"]
69 | Error["EncodingServices"]
70 | R
71 | Scope
72 | WorkflowEngine
73 | WorkflowInstance
74 >
75 readonly executeEncoded: Effect.Effect<
76 unknown,
77 unknown,
78 | Success["DecodingServices"]
79 | Success["EncodingServices"]
80 | Error["DecodingServices"]
81 | Error["EncodingServices"]
82 | R
83 | Scope
84 | WorkflowEngine
85 | WorkflowInstance
86 >
87}
88
89/**
90 * Type-erased activity shape for APIs that only need the activity identity,
91 * name, annotations, and encoded execution.
92 *
93 * @stability unstable
94 * @category models
95 * @since 4.0.0
96 */
97export interface Any {
98 readonly [TypeId]: typeof TypeId
99 readonly name: string
100 readonly executeEncoded: Effect.Effect<any, any, any>
101 readonly annotations: Context.Context<never>
102}
103
104/**
105 * Type-erased activity shape that also exposes success and error schemas for
106 * derived workflow APIs.
107 *
108 * @stability unstable
109 * @category models
110 * @since 4.0.0
111 */
112export interface AnyWithProps {
113 readonly [TypeId]: typeof TypeId
114 readonly name: string
115 readonly successSchema: Schema.Top
116 readonly errorSchema: Schema.Top
117 readonly executeEncoded: Effect.Effect<any, any, any>
118}
119
120/**
121 * Creates a workflow activity from an effect, using the provided schemas to
122 * encode successes and failures for durable execution.
123 *
124 * **Gotchas**
125 *
126 * Only completed activity results are memoized. If the activity suspends while
127 * awaiting child workflows or a durable clock, its body runs again when the
128 * parent workflow replays. Side effects before the suspension can repeat;
129 * make those side effects idempotent.
130 *
131 * @stability unstable
132 * @category constructors
133 * @since 4.0.0
134 */
135export const make = <
136 R,
137 Success extends Schema.Constraint = Schema.Void,
138 Error extends Schema.Constraint = Schema.Never
139>(options: {
140 readonly name: string
141 readonly success?: Success | undefined
142 readonly error?: Error | undefined
143 readonly execute: Effect.Effect<Success["Type"], Error["Type"], R>
144 readonly interruptRetryPolicy?: Schedule.Schedule<any, Cause.Cause<unknown>> | undefined
145 readonly annotations?: Context.Context<never> | undefined
146}): Activity<Success, Error, Exclude<R, WorkflowInstance | WorkflowEngine | Scope>> => {
147 const successSchema = options.success ?? (Schema.Void as any as Success)
148 const errorSchema = options.error ?? (Schema.Never as any as Error)
149 const successSchemaJson = Schema.toCodecJson(successSchema)
150 const errorSchemaJson = Schema.toCodecJson(errorSchema)
151 // oxlint-disable-next-line prefer-const
152 let execute!: Effect.Effect<Success["Type"], Error["Type"], any>
153 const executeWithoutInterrupt = retryOnInterrupt(
154 options.name,
155 options.interruptRetryPolicy
156 )(options.execute)
157 const self: Activity<Success, Error, Exclude<R, WorkflowInstance | WorkflowEngine>> = {
158 ...Effectable.Prototype<Activity<Success, Error, R>>({
159 label: "Activity",
160 evaluate(_) {
161 return execute
162 }
163 }),
164 [TypeId]: TypeId,
165 name: options.name,
166 successSchema,
167 errorSchema,
168 exitSchema: Schema.Exit(successSchemaJson, errorSchemaJson, Schema.Defect()),
169 exitSchemaPartial: Schema.Exit(successSchemaJson, errorSchemaJson, Schema.Unknown),
170 annotations: options.annotations ?? Context.empty(),
171 annotate(tag: Context.Key<any, any>, value: any) {
172 return make({
173 ...options,
174 annotations: Context.add(self.annotations, tag, value)
175 })
176 },
177 annotateMerge(context: Context.Context<any>) {
178 return make({
179 ...options,
180 annotations: Context.merge(self.annotations, context)
181 })
182 },
183 execute: executeWithoutInterrupt,
184 executeEncoded: Effect.matchEffect(executeWithoutInterrupt, {
185 onFailure: (error) => Effect.flatMap(Effect.orDie(Schema.encodeEffect(errorSchemaJson)(error)), Effect.fail),
186 onSuccess: (value) => Effect.orDie(Schema.encodeEffect(successSchemaJson)(value))
187 })
188 } as any
189 execute = makeExecute(self)
190 return self
191}
192
193const interruptRetryPolicy = Schedule.min([
194 Schedule.exponential(400, 1.5),
195 Schedule.spaced("10 seconds")
196]).pipe(
197 Schedule.setInputType<Cause.Cause<unknown>>(),
198 Schedule.while((meta) => meta.attempt <= 10 && Cause.hasInterrupts(meta.input))
199)
200
201const retryOnInterrupt = (
202 name: string,
203 policy: Schedule.Schedule<any, Cause.Cause<unknown>> = interruptRetryPolicy
204) =>
205<A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R | WorkflowInstance> =>
206 Effect.flatMap(InstanceTag, (instance) =>
207 effect.pipe(
208 Effect.sandbox,
209 // A suspension interrupts the activity body on purpose and must surface
210 // as a suspended result instead of being retried.
211 Effect.retry({ schedule: policy, while: () => !instance.suspended }),
212 Effect.catch((cause) =>
213 Cause.hasInterrupts(cause) && !instance.suspended
214 ? Effect.die(`Activity "${name}" interrupted and retry attempts exhausted`)
215 : Effect.failCause(cause)
216 )
217 ))
218
219/**
220 * Retries an effect with `Effect.retry` while updating `CurrentAttempt` for
221 * each attempt.
222 *
223 * @stability unstable
224 * @category error handling
225 * @since 4.0.0
226 */
227export const retry: {
228 <E, O extends Types.NoExcessProperties<Omit<Effect.Retry.Options<E>, "schedule">, O>>(
229 options: O
230 ): <A, R>(self: Effect.Effect<A, E, R>) => Effect.Retry.Return<R, E, A, O>
231 <A, E, R, O extends Types.NoExcessProperties<Omit<Effect.Retry.Options<E>, "schedule">, O>>(
232 self: Effect.Effect<A, E, R>,
233 options: O
234 ): Effect.Retry.Return<R, E, A, O>
235} = dual(
236 2,
237 (effect: Effect.Effect<any, any, any>, options: {}) =>
238 Effect.suspend(() => {
239 let attempt = 1
240 return Effect.suspend(() => Effect.provideService(effect, CurrentAttempt, attempt++)).pipe(Effect.retry(options))
241 })
242)
243
244/**
245 * Context reference containing the current activity retry attempt, defaulting
246 * to `1`.
247 *
248 * @stability unstable
249 * @category services
250 * @since 4.0.0
251 */
252export const CurrentAttempt = Context.Reference<number>(
253 "effect/workflow/Activity/CurrentAttempt",
254 { defaultValue: () => 1 }
255)
256
257/**
258 * Computes a deterministic activity idempotency key from the current workflow
259 * execution ID, the supplied name, and optionally the current attempt.
260 *
261 * @stability unstable
262 * @category idempotency
263 * @since 4.0.0
264 */
265export const idempotencyKey: (
266 name: string,
267 options?: {
268 readonly includeAttempt?: boolean | undefined
269 } | undefined
270) => Effect.Effect<string, never, WorkflowInstance> = Effect.fnUntraced(function*(name: string, options?: {
271 readonly includeAttempt?: boolean | undefined
272}) {
273 const instance = yield* InstanceTag
274 let key = `${instance.executionId}`
275 if (options?.includeAttempt) {
276 const attempt = yield* CurrentAttempt
277 key += `-${attempt}`
278 }
279 key += `-${name}`
280 return yield* makeHashDigest(key)
281})
282
283/**
284 * Runs a non-empty collection of activities as a durable race and returns the
285 * first completed success or failure using unioned success and error schemas.
286 *
287 * @stability unstable
288 * @category racing
289 * @since 4.0.0
290 */
291export const raceAll = <const Activities extends NonEmptyReadonlyArray<Any>>(
292 name: string,
293 activities: Activities
294): Effect.Effect<
295 Activities[number] extends Activity<infer _A, infer _E, infer _R> ? _A["Type"] : never,
296 Activities[number] extends Activity<infer _A, infer _E, infer _R> ? _E["Type"] : never,
297 | (Activities[number] extends Activity<infer Success, infer Error, infer R>
298 ? Success["DecodingServices"] | Error["DecodingServices"] | R
299 : never)
300 | WorkflowEngine
301 | WorkflowInstance
302> =>
303 DurableDeferred.raceAll({
304 name: `Activity/${name}`,
305 success: Schema.Union(
306 activities.map((activity) => (activity as any).successSchema)
307 ),
308 error: Schema.Union(
309 activities.map((activity) => (activity as any).errorSchema)
310 ),
311 effects: activities.map((activity) => (activity as any)) as any
312 }) as any
313
314// -----------------------------------------------------------------------------
315// internal
316// -----------------------------------------------------------------------------
317
318const EngineTag = Context.Service<WorkflowEngine, WorkflowEngine["Service"]>(
319 "effect/workflow/WorkflowEngine" satisfies typeof WorkflowEngine.key
320)
321const InstanceTag = Context.Service<WorkflowInstance, WorkflowInstance["Service"]>(
322 "effect/workflow/WorkflowEngine/WorkflowInstance" satisfies typeof WorkflowInstance.key
323)
324
325const makeExecute = Effect.fnUntraced(function*<
326 R,
327 Success extends Schema.Constraint = typeof Schema.Void,
328 Error extends Schema.Constraint = typeof Schema.Never
329>(activity: Activity<Success, Error, R>) {
330 const engine = yield* EngineTag
331 const instance = yield* InstanceTag
332 const attempt = yield* CurrentAttempt
333 yield* Effect.annotateCurrentSpan({ executionId: instance.executionId })
334 const result = yield* Workflow.wrapActivityResult(
335 engine.activityExecute(activity, attempt),
336 (_) => _._tag === "Suspended"
337 )
338 if (result._tag === "Suspended") {
339 return yield* Workflow.suspend(instance)
340 }
341 return yield* result.exit
342}, (effect, activity) =>
343 Effect.withSpan(effect, activity.name, {
344 captureStackTrace: false
345 }))