1/**
2 * Durable timers for workflow sleeps.
3 *
4 * `make` creates a `DurableClock` with a name, duration, and deferred wake-up
5 * signal. `sleep` ignores zero durations, runs short sleeps through an
6 * in-memory activity, and schedules longer sleeps through the `WorkflowEngine`
7 * before awaiting the durable deferred tied to the clock.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import * as Context from "../Context.ts"
13import * as Duration from "../Duration.ts"
14import * as Effect from "../Effect.ts"
15import type * as Schema from "../Schema.ts"
16import * as Activity from "./Activity.ts"
17import * as DurableDeferred from "./DurableDeferred.ts"
18import type { WorkflowEngine, WorkflowInstance } from "./WorkflowEngine.ts"
19
20const TypeId = "~effect/workflow/DurableClock"
21
22/**
23 * Represents a durable workflow timer with a name, duration, and deferred
24 * completed when the timer wakes.
25 *
26 * @stability unstable
27 * @category models
28 * @since 4.0.0
29 */
30export interface DurableClock {
31 readonly [TypeId]: typeof TypeId
32 readonly name: string
33 readonly duration: Duration.Duration
34 readonly deferred: DurableDeferred.DurableDeferred<typeof Schema.Void>
35}
36
37/**
38 * Creates a durable clock definition and its associated deferred wake-up
39 * signal.
40 *
41 * @stability unstable
42 * @category constructors
43 * @since 4.0.0
44 */
45export const make = (options: {
46 readonly name: string
47 readonly duration: Duration.Input
48}): DurableClock => ({
49 [TypeId]: TypeId,
50 name: options.name,
51 duration: Duration.fromInputUnsafe(options.duration),
52 deferred: DurableDeferred.make(`DurableClock/${options.name}`)
53})
54
55const EngineTag = Context.Service<WorkflowEngine, WorkflowEngine["Service"]>(
56 "effect/workflow/WorkflowEngine" satisfies typeof WorkflowEngine.key
57)
58
59const InstanceTag = Context.Service<
60 WorkflowInstance,
61 WorkflowInstance["Service"]
62>(
63 "effect/workflow/WorkflowEngine/WorkflowInstance" satisfies typeof WorkflowInstance.key
64)
65
66/**
67 * Waits inside a workflow, using an in-memory activity for durations at or
68 * below the threshold and scheduling a durable clock for longer durations.
69 *
70 * @stability unstable
71 * @category delays & timeouts
72 * @since 4.0.0
73 */
74export const sleep: (
75 options: {
76 readonly name: string
77 readonly duration: Duration.Input
78 /**
79 * If the duration is less than or equal to this threshold, the clock will
80 * be executed in memory.
81 *
82 * @default 60 seconds
83 */
84 readonly inMemoryThreshold?: Duration.Input | undefined
85 }
86) => Effect.Effect<
87 void,
88 never,
89 WorkflowEngine | WorkflowInstance
90> = Effect.fnUntraced(function*(options: {
91 readonly name: string
92 readonly duration: Duration.Input
93 readonly inMemoryThreshold?: Duration.Input | undefined
94}) {
95 const duration = Duration.fromInputUnsafe(options.duration)
96 if (Duration.isZero(duration)) {
97 return
98 }
99
100 const inMemoryThreshold = options.inMemoryThreshold !== undefined
101 ? Duration.fromInputUnsafe(options.inMemoryThreshold)
102 : defaultInMemoryThreshold
103
104 if (Duration.isLessThanOrEqualTo(duration, inMemoryThreshold)) {
105 return yield* Activity.make({
106 name: `DurableClock/${options.name}`,
107 execute: Effect.sleep(duration)
108 })
109 }
110
111 const engine = yield* EngineTag
112 const instance = yield* InstanceTag
113 const clock = make(options)
114 yield* engine.scheduleClock(instance.workflow, {
115 executionId: instance.executionId,
116 clock
117 })
118 return yield* DurableDeferred.await(clock.deferred)
119})
120
121const defaultInMemoryThreshold = Duration.seconds(60)