1/**
2 * Defines named wait points for durable workflow executions.
3 *
4 * A `DurableDeferred` has a stable name and schemas for the value that will be
5 * recorded later. Workflows can await it, suspend when no result exists yet, and
6 * resume after its result is recorded. Tokens identify the workflow name,
7 * execution id, and deferred name so external code can complete the correct
8 * wait point later.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Arr from "../Array.ts"
14import type { NonEmptyReadonlyArray } from "../Array.ts"
15import type * as Brand from "../Brand.ts"
16import * as Cause from "../Cause.ts"
17import * as Context from "../Context.ts"
18import * as Effect from "../Effect.ts"
19import * as Base64Url from "../encoding/Base64Url.ts"
20import * as Exit from "../Exit.ts"
21import * as Filter from "../Filter.ts"
22import { dual } from "../Function.ts"
23import * as Option from "../Option.ts"
24import * as Schema from "../Schema.ts"
25import * as SchemaGetter from "../SchemaGetter.ts"
26import type * as Activity from "./Activity.ts"
27import * as Workflow from "./Workflow.ts"
28import type { WorkflowEngine, WorkflowInstance } from "./WorkflowEngine.ts"
29
30const TypeId = "~effect/workflow/DurableDeferred"
31
32/**
33 * Named durable deferred value whose completion is persisted by the workflow
34 * engine and encoded with success and error schemas.
35 *
36 * @stability unstable
37 * @category models
38 * @since 4.0.0
39 */
40export interface DurableDeferred<
41 Success extends Schema.Constraint,
42 Error extends Schema.Constraint = Schema.Never
43> {
44 readonly [TypeId]: typeof TypeId
45 readonly name: string
46 readonly successSchema: Success
47 readonly errorSchema: Error
48 readonly exitSchema: Schema.Exit<Schema.Top, Schema.Top, Schema.Top>
49 readonly withActivityAttempt: Effect.Effect<DurableDeferred<Success, Error>>
50}
51
52/**
53 * Type-erased durable deferred shape for APIs that only need the deferred
54 * identity and name.
55 *
56 * @stability unstable
57 * @category models
58 * @since 4.0.0
59 */
60export interface Any {
61 readonly [TypeId]: typeof TypeId
62 readonly name: string
63}
64
65/**
66 * Type-erased durable deferred shape that also exposes success, error, and
67 * exit schemas.
68 *
69 * @stability unstable
70 * @category models
71 * @since 4.0.0
72 */
73export interface AnyWithProps {
74 readonly [TypeId]: typeof TypeId
75 readonly name: string
76 readonly successSchema: Schema.Top
77 readonly errorSchema: Schema.Top
78 readonly exitSchema: Schema.Exit<any, any, any>
79}
80
81/**
82 * Creates a named durable deferred with optional success and error schemas for
83 * persisted completion.
84 *
85 * @stability unstable
86 * @category constructors
87 * @since 4.0.0
88 */
89export const make = <
90 Success extends Schema.Constraint = Schema.Void,
91 Error extends Schema.Constraint = Schema.Never
92>(
93 name: string,
94 options?: {
95 readonly success?: Success | undefined
96 readonly error?: Error | undefined
97 }
98): DurableDeferred<Success, Error> => {
99 const successSchema = options?.success ?? (Schema.Void as any as Success)
100 const errorSchema = options?.error ?? (Schema.Never as any as Error)
101 return {
102 [TypeId]: TypeId as typeof TypeId,
103 name,
104 successSchema,
105 errorSchema,
106 exitSchema: Schema.Exit(
107 Schema.toCodecJson(successSchema),
108 Schema.toCodecJson(errorSchema),
109 Schema.toCodecJson(Schema.Defect())
110 ) as any,
111 withActivityAttempt: Effect.gen(function*() {
112 const attempt = yield* CurrentAttempt
113 return make(`${name}/${attempt}`, {
114 success: successSchema,
115 error: errorSchema
116 })
117 })
118 }
119}
120
121const EngineTag = Context.Service<WorkflowEngine, WorkflowEngine["Service"]>(
122 "effect/workflow/WorkflowEngine" satisfies typeof WorkflowEngine.key
123)
124
125const InstanceTag = Context.Service<
126 WorkflowInstance,
127 WorkflowInstance["Service"]
128>(
129 "effect/workflow/WorkflowEngine/WorkflowInstance" satisfies typeof WorkflowInstance.key
130)
131
132const CurrentAttempt = Context.Reference<number>(
133 "effect/workflow/Activity/CurrentAttempt" satisfies typeof Activity.CurrentAttempt.key,
134 { defaultValue: () => 1 }
135)
136
137const await_: <Success extends Schema.Constraint, Error extends Schema.Constraint>(
138 self: DurableDeferred<Success, Error>
139) => Effect.Effect<
140 Success["Type"],
141 Error["Type"],
142 | WorkflowEngine
143 | WorkflowInstance
144 | Success["DecodingServices"]
145 | Error["DecodingServices"]
146> = Effect.fnUntraced(function*<
147 Success extends Schema.Constraint,
148 Error extends Schema.Constraint
149>(self: DurableDeferred<Success, Error>) {
150 const engine = yield* EngineTag
151 const instance = yield* InstanceTag
152 // Register before the read so any later completion can preempt the run.
153 instance.awaitedDeferreds.add(self.name)
154 const exit = yield* Workflow.wrapActivityResult(
155 engine.deferredResult(self),
156 Option.isNone
157 )
158 if (Option.isNone(exit)) {
159 return yield* Workflow.suspend(instance)
160 }
161 return yield* exit.value as Exit.Exit<any, any>
162})
163
164/**
165 * @stability unstable
166 */
167export {
168 /**
169 * Waits for the durable deferred, suspending the current workflow when no
170 * persisted completion is available.
171 *
172 * @stability unstable
173 * @category combinators
174 * @since 4.0.0
175 */
176 await_ as await
177}
178
179/**
180 * Runs an effect and records its exit into the durable deferred, resuming
181 * workflows that are waiting on that deferred.
182 *
183 * @stability unstable
184 * @category combinators
185 * @since 4.0.0
186 */
187export const into: {
188 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
189 self: DurableDeferred<Success, Error>
190 ): <R>(
191 effect: Effect.Effect<Success["Type"], Error["Type"], R>
192 ) => Effect.Effect<
193 Success["Type"],
194 Error["Type"],
195 | R
196 | WorkflowEngine
197 | WorkflowInstance
198 | Success["DecodingServices"]
199 | Error["DecodingServices"]
200 | Success["EncodingServices"]
201 | Error["EncodingServices"]
202 >
203 <Success extends Schema.Constraint, Error extends Schema.Constraint, R>(
204 effect: Effect.Effect<Success["Type"], Error["Type"], R>,
205 self: DurableDeferred<Success, Error>
206 ): Effect.Effect<
207 Success["Type"],
208 Error["Type"],
209 | R
210 | WorkflowEngine
211 | WorkflowInstance
212 | Success["DecodingServices"]
213 | Error["DecodingServices"]
214 | Success["EncodingServices"]
215 | Error["EncodingServices"]
216 >
217} = dual(
218 2,
219 <Success extends Schema.Constraint, Error extends Schema.Constraint, R>(
220 effect: Effect.Effect<Success["Type"], Error["Type"], R>,
221 self: DurableDeferred<Success, Error>
222 ): Effect.Effect<
223 Success["Type"],
224 Error["Type"],
225 | R
226 | WorkflowEngine
227 | WorkflowInstance
228 | Success["DecodingServices"]
229 | Error["DecodingServices"]
230 | Success["EncodingServices"]
231 | Error["EncodingServices"]
232 > =>
233 Effect.contextWith(
234 (context: Context.Context<WorkflowEngine | WorkflowInstance>) => {
235 const engine = Context.get(context, EngineTag)
236 const parentInstance = Context.get(context, InstanceTag)
237 const instance = { ...parentInstance }
238 return Effect.onExit(
239 Effect.provideService(effect, InstanceTag, instance),
240 Effect.fnUntraced(function*(exit) {
241 if (Exit.isFailure(exit)) {
242 const [interrupts, reasons] = Arr.partition(
243 exit.cause.reasons,
244 Filter.fromPredicate(Cause.isInterruptReason)
245 )
246 if (interrupts.length === exit.cause.reasons.length) {
247 // An interrupt-only exit is never a result: the effect was
248 // suspended, preempted or interrupted, so record nothing.
249 if (instance.suspended) {
250 parentInstance.suspended = true
251 }
252 return
253 } else if (interrupts.length > 0) {
254 exit = Exit.failCause(Cause.fromReasons(reasons))
255 }
256 }
257 yield* engine.deferredDone(self, {
258 workflowName: instance.workflow._tag,
259 executionId: instance.executionId,
260 deferredName: self.name,
261 exit
262 })
263 })
264 )
265 }
266 )
267)
268
269/**
270 * Runs effects as a durable race, returning a previously persisted result when
271 * present or completing a named deferred with the first result.
272 *
273 * @stability unstable
274 * @category racing
275 * @since 4.0.0
276 */
277export const raceAll = <
278 const Effects extends NonEmptyReadonlyArray<Effect.Effect<any, any, any>>,
279 Success extends Schema.Schema<Effect.Success<Effects[number]>>,
280 Error extends Schema.Schema<Effect.Error<Effects[number]>>
281>(options: {
282 name: string
283 success: Success
284 error: Error
285 effects: Effects
286}): Effect.Effect<
287 Effect.Success<Effects[number]>,
288 Effect.Error<Effects[number]>,
289 | Effect.Services<Effects[number]>
290 | Success["DecodingServices"]
291 | Success["EncodingServices"]
292 | Error["DecodingServices"]
293 | Error["EncodingServices"]
294 | WorkflowEngine
295 | WorkflowInstance
296> => {
297 const deferred = make<any, any>(`raceAll/${options.name}`, {
298 success: options.success,
299 error: options.error
300 })
301 return Effect.gen(function*() {
302 const engine = yield* EngineTag
303 const exit = yield* engine.deferredResult(deferred)
304 if (Option.isSome(exit)) {
305 return yield* exit.value
306 }
307 return yield* into(
308 Effect.raceAll(options.effects),
309 deferred
310 )
311 })
312}
313
314/**
315 * Runtime brand identifier for durable deferred tokens.
316 *
317 * @stability unstable
318 * @category type IDs
319 * @since 4.0.0
320 */
321export const TokenTypeId = "~effect/workflow/DurableDeferred/Token"
322
323/**
324 * Type-level brand identifier for `Token` values.
325 *
326 * @stability unstable
327 * @category type IDs
328 * @since 4.0.0
329 */
330export type TokenTypeId = typeof TokenTypeId
331
332/**
333 * Branded string token identifying a durable deferred for a workflow
334 * execution.
335 *
336 * @stability unstable
337 * @category models
338 * @since 4.0.0
339 */
340export type Token = Brand.Branded<string, TokenTypeId>
341
342/**
343 * Schema for branded durable deferred tokens.
344 *
345 * @stability unstable
346 * @category schemas
347 * @since 4.0.0
348 */
349export const Token: Schema.brand<Schema.String, TokenTypeId> = Schema.String.pipe(Schema.brand(TokenTypeId))
350
351/**
352 * Schema for a decoded durable deferred token containing the workflow
353 * name, execution ID, and deferred name.
354 *
355 * @stability unstable
356 * @category schemas
357 * @since 4.0.0
358 */
359export class TokenParsed extends Schema.Class<TokenParsed>(
360 "effect/workflow/DurableDeferred/TokenParsed"
361)({
362 workflowName: Schema.String,
363 executionId: Schema.String,
364 deferredName: Schema.String
365}) {
366 /**
367 * Encodes the parsed workflow, execution, and deferred names back into a token.
368 *
369 * @stability unstable
370 * @since 4.0.0
371 */
372 get asToken(): Token {
373 return Base64Url.encode(
374 JSON.stringify([this.workflowName, this.executionId, this.deferredName])
375 ) as Token
376 }
377
378 /**
379 * Schema for decoding and encoding durable deferred tokens as strings.
380 *
381 * @stability unstable
382 * @since 4.0.0
383 */
384 static readonly FromString = Schema.String.pipe(
385 Schema.decodeTo(
386 Schema.fromJsonString(
387 Schema.Tuple([Schema.String, Schema.String, Schema.String])
388 ),
389 {
390 decode: SchemaGetter.decodeBase64UrlString(),
391 encode: SchemaGetter.encodeBase64Url()
392 }
393 ),
394 Schema.decodeTo(TokenParsed, {
395 decode: SchemaGetter.transform(
396 ([workflowName, executionId, deferredName]) =>
397 new TokenParsed({
398 workflowName,
399 executionId,
400 deferredName
401 })
402 ),
403 encode: SchemaGetter.transform(
404 (parsed) =>
405 [
406 parsed.workflowName,
407 parsed.executionId,
408 parsed.deferredName
409 ] as const
410 )
411 })
412 )
413
414 /**
415 * Decodes a durable deferred token string into its parsed components.
416 *
417 * @stability unstable
418 * @since 4.0.0
419 */
420 static readonly fromString = Schema.decodeSync(TokenParsed.FromString)
421
422 /**
423 * Encodes parsed durable deferred token components into a token string.
424 *
425 * @stability unstable
426 * @since 4.0.0
427 */
428 static readonly encode = Schema.encodeSync(TokenParsed.FromString)
429}
430
431/**
432 * Creates a token for a durable deferred using the current workflow instance's
433 * workflow name and execution ID.
434 *
435 * @stability unstable
436 * @category constructors
437 * @since 4.0.0
438 */
439export const token: <Success extends Schema.Constraint, Error extends Schema.Constraint>(
440 self: DurableDeferred<Success, Error>
441) => Effect.Effect<Token, never, WorkflowInstance> = Effect.fnUntraced(
442 function*<Success extends Schema.Constraint, Error extends Schema.Constraint>(
443 self: DurableDeferred<Success, Error>
444 ) {
445 const instance = yield* InstanceTag
446 return tokenFromExecutionId(self, instance)
447 }
448)
449
450/**
451 * Creates a durable deferred token from an explicit workflow, execution ID,
452 * and deferred name.
453 *
454 * @stability unstable
455 * @category constructors
456 * @since 4.0.0
457 */
458export const tokenFromExecutionId: {
459 (options: {
460 readonly workflow: Workflow.Any
461 readonly executionId: string
462 }): <Success extends Schema.Constraint, Error extends Schema.Constraint>(
463 self: DurableDeferred<Success, Error>
464 ) => Token
465 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
466 self: DurableDeferred<Success, Error>,
467 options: { readonly workflow: Workflow.Any; readonly executionId: string }
468 ): Token
469} = dual(
470 2,
471 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
472 self: DurableDeferred<Success, Error>,
473 options: {
474 readonly workflow: Workflow.Any
475 readonly executionId: string
476 }
477 ): Token =>
478 new TokenParsed({
479 workflowName: options.workflow._tag,
480 executionId: options.executionId,
481 deferredName: self.name
482 }).asToken
483)
484
485/**
486 * Creates a durable deferred token by deriving the workflow execution ID from
487 * the supplied workflow payload.
488 *
489 * @stability unstable
490 * @category constructors
491 * @since 4.0.0
492 */
493export const tokenFromPayload: {
494 <W extends Workflow.Any>(options: {
495 readonly workflow: W
496 readonly payload: Workflow.PayloadSchema<W>["~type.make.in"]
497 }): <Success extends Schema.Constraint, Error extends Schema.Constraint>(
498 self: DurableDeferred<Success, Error>
499 ) => Effect.Effect<Token>
500 <
501 Success extends Schema.Constraint,
502 Error extends Schema.Constraint,
503 W extends Workflow.Any
504 >(
505 self: DurableDeferred<Success, Error>,
506 options: {
507 readonly workflow: W
508 readonly payload: Workflow.PayloadSchema<W>["~type.make.in"]
509 }
510 ): Effect.Effect<Token>
511} = dual(
512 2,
513 <
514 Success extends Schema.Constraint,
515 Error extends Schema.Constraint,
516 W extends Workflow.Any
517 >(
518 self: DurableDeferred<Success, Error>,
519 options: {
520 readonly workflow: W
521 readonly payload: Workflow.PayloadSchema<W>["~type.make.in"]
522 }
523 ): Effect.Effect<Token> =>
524 Effect.map(options.workflow.executionId(options.payload), (executionId) =>
525 tokenFromExecutionId(self, {
526 workflow: options.workflow,
527 executionId
528 }))
529)
530
531/**
532 * Completes the durable deferred identified by a token with the supplied exit,
533 * encoding the result through the deferred schemas.
534 *
535 * @stability unstable
536 * @category combinators
537 * @since 4.0.0
538 */
539export const done: {
540 <Success extends Schema.Constraint, Error extends Schema.Constraint>(options: {
541 readonly token: Token
542 readonly exit: Exit.Exit<Success["Type"], Error["Type"]>
543 }): (
544 self: DurableDeferred<Success, Error>
545 ) => Effect.Effect<
546 void,
547 never,
548 WorkflowEngine | Success["EncodingServices"] | Error["EncodingServices"]
549 >
550 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
551 self: DurableDeferred<Success, Error>,
552 options: {
553 readonly token: Token
554 readonly exit: Exit.Exit<Success["Type"], Error["Type"]>
555 }
556 ): Effect.Effect<
557 void,
558 never,
559 WorkflowEngine | Success["EncodingServices"] | Error["EncodingServices"]
560 >
561} = dual(
562 2,
563 Effect.fnUntraced(function*<
564 Success extends Schema.Constraint,
565 Error extends Schema.Constraint
566 >(
567 self: DurableDeferred<Success, Error>,
568 options: {
569 readonly token: Token
570 readonly exit: Exit.Exit<Success["Type"], Error["Type"]>
571 }
572 ) {
573 const engine = yield* EngineTag
574 const token = TokenParsed.fromString(options.token)
575 yield* engine.deferredDone(self, {
576 workflowName: token.workflowName,
577 executionId: token.executionId,
578 deferredName: token.deferredName,
579 exit: options.exit
580 })
581 })
582)
583
584/**
585 * Completes the durable deferred identified by a token with a successful
586 * value.
587 *
588 * @stability unstable
589 * @category combinators
590 * @since 4.0.0
591 */
592export const succeed: {
593 <Success extends Schema.Constraint, Error extends Schema.Constraint>(options: {
594 readonly token: Token
595 readonly value: Success["Type"]
596 }): (
597 self: DurableDeferred<Success, Error>
598 ) => Effect.Effect<void, never, WorkflowEngine | Success["EncodingServices"]>
599 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
600 self: DurableDeferred<Success, Error>,
601 options: {
602 readonly token: Token
603 readonly value: Success["Type"]
604 }
605 ): Effect.Effect<void, never, WorkflowEngine | Success["EncodingServices"]>
606} = dual(
607 2,
608 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
609 self: DurableDeferred<Success, Error>,
610 options: {
611 readonly token: Token
612 readonly value: Success["Type"]
613 }
614 ): Effect.Effect<void, never, WorkflowEngine | Success["EncodingServices"]> =>
615 done(self, {
616 token: options.token,
617 exit: Exit.succeed(options.value)
618 })
619)
620
621/**
622 * Completes the durable deferred identified by a token with a typed failure.
623 *
624 * @stability unstable
625 * @category combinators
626 * @since 4.0.0
627 */
628export const fail: {
629 <Success extends Schema.Constraint, Error extends Schema.Constraint>(options: {
630 readonly token: Token
631 readonly error: Error["Type"]
632 }): (
633 self: DurableDeferred<Success, Error>
634 ) => Effect.Effect<void, never, WorkflowEngine | Error["EncodingServices"]>
635 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
636 self: DurableDeferred<Success, Error>,
637 options: {
638 readonly token: Token
639 readonly error: Error["Type"]
640 }
641 ): Effect.Effect<void, never, WorkflowEngine | Error["EncodingServices"]>
642} = dual(
643 2,
644 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
645 self: DurableDeferred<Success, Error>,
646 options: {
647 readonly token: Token
648 readonly error: Error["Type"]
649 }
650 ): Effect.Effect<void, never, WorkflowEngine | Error["EncodingServices"]> =>
651 done(self, {
652 token: options.token,
653 exit: Exit.fail(options.error)
654 })
655)
656
657/**
658 * Completes the durable deferred identified by a token with a failure cause.
659 *
660 * @stability unstable
661 * @category combinators
662 * @since 4.0.0
663 */
664export const failCause: {
665 <Success extends Schema.Constraint, Error extends Schema.Constraint>(options: {
666 readonly token: Token
667 readonly cause: Cause.Cause<Error["Type"]>
668 }): (
669 self: DurableDeferred<Success, Error>
670 ) => Effect.Effect<void, never, WorkflowEngine | Error["EncodingServices"]>
671 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
672 self: DurableDeferred<Success, Error>,
673 options: {
674 readonly token: Token
675 readonly cause: Cause.Cause<Error["Type"]>
676 }
677 ): Effect.Effect<void, never, WorkflowEngine | Error["EncodingServices"]>
678} = dual(
679 2,
680 <Success extends Schema.Constraint, Error extends Schema.Constraint>(
681 self: DurableDeferred<Success, Error>,
682 options: {
683 readonly token: Token
684 readonly cause: Cause.Cause<Error["Type"]>
685 }
686 ): Effect.Effect<void, never, WorkflowEngine | Error["EncodingServices"]> =>
687 done(self, {
688 token: options.token,
689 exit: Exit.failCause(options.cause)
690 })
691)