1import * as Duration from "../Duration.ts"
2import type { Effect } from "../Effect.ts"
3import * as Api from "../ExecutionPlan.ts"
4import type * as Exit from "../Exit.ts"
5import { dual, identity } from "../Function.ts"
6import * as Result from "../Result.ts"
7import * as Schedule from "../Schedule.ts"
8import { isEffect } from "./core.ts"
9import * as effect from "./effect.ts"
10import * as internalLayer from "./layer.ts"
11import * as internalSchedule from "./schedule.ts"
12
13/** @internal */
14export interface AttemptState {
15 readonly attempt: number
16 readonly stepAttempt: number
17 readonly stepIndex: number
18 readonly startNanos: bigint
19}
20
21/** @internal */
22export interface EventEmitter {
23 readonly begin: Effect<AttemptState, never, any>
24 readonly end: (state: AttemptState, exit: Exit.Exit<any, any>) => Effect<void, never, any>
25}
26
27/** @internal */
28export const makeEventEmitter = (
29 onEvent: (event: Api.Event<any>) => Effect<void, never, any>,
30 currentMetadata: () => Api.Metadata
31): EventEmitter => {
32 let lastStepIndex = -1
33 let stepAttempt = 0
34 const emit = (event: Api.Event<any>) => effect.ignoreCause(onEvent(event))
35 return {
36 begin: effect.clockWith((clock) =>
37 effect.suspend(() => {
38 const meta = currentMetadata()
39 if (meta.stepIndex !== lastStepIndex) {
40 lastStepIndex = meta.stepIndex
41 stepAttempt = 0
42 }
43 stepAttempt++
44 const state: AttemptState = {
45 attempt: meta.attempt,
46 stepAttempt,
47 stepIndex: meta.stepIndex,
48 startNanos: clock.monotonicTimeNanosUnsafe()
49 }
50 return effect.as(
51 emit({
52 _tag: "AttemptStart",
53 attempt: state.attempt,
54 stepAttempt: state.stepAttempt,
55 stepIndex: state.stepIndex
56 }),
57 state
58 )
59 })
60 ),
61 end: (state, exit) =>
62 effect.clockWith((clock) => {
63 const duration = Duration.nanos(clock.monotonicTimeNanosUnsafe() - state.startNanos)
64 return emit(
65 exit._tag === "Success"
66 ? {
67 _tag: "AttemptSuccess",
68 attempt: state.attempt,
69 stepAttempt: state.stepAttempt,
70 stepIndex: state.stepIndex,
71 duration
72 }
73 : {
74 _tag: "AttemptFailure",
75 attempt: state.attempt,
76 stepAttempt: state.stepAttempt,
77 stepIndex: state.stepIndex,
78 duration,
79 cause: exit.cause
80 }
81 )
82 })
83 }
84}
85
86/** @internal */
87export const withExecutionPlan: {
88 <Input, Provides, PlanE, PlanR, RX = never>(
89 plan: Api.ExecutionPlan<{
90 provides: Provides
91 input: Input
92 error: PlanE
93 requirements: PlanR
94 }>,
95 options?: {
96 readonly onEvent?: ((event: Api.Event<Input | PlanE>) => Effect<void, never, RX>) | undefined
97 }
98 ): <A, E extends Input, R>(effect: Effect<A, E, R>) => Effect<
99 A,
100 E | PlanE,
101 Exclude<R, Provides> | PlanR | RX
102 >
103 <A, E extends Input, R, Provides, Input, PlanE, PlanR, RX = never>(
104 effect: Effect<A, E, R>,
105 plan: Api.ExecutionPlan<{
106 provides: Provides
107 input: Input
108 error: PlanE
109 requirements: PlanR
110 }>,
111 options?: {
112 readonly onEvent?: ((event: Api.Event<E | PlanE>) => Effect<void, never, RX>) | undefined
113 }
114 ): Effect<
115 A,
116 E | PlanE,
117 Exclude<R, Provides> | PlanR | RX
118 >
119} = dual((args) => isEffect(args[0]), <A, E extends Input, R, Provides, Input, PlanE, PlanR, RX>(
120 self: Effect<A, E, R>,
121 plan: Api.ExecutionPlan<{
122 provides: Provides
123 input: Input
124 error: PlanE
125 requirements: PlanR
126 }>,
127 options?: {
128 readonly onEvent?: ((event: Api.Event<E | PlanE>) => Effect<void, never, RX>) | undefined
129 }
130) =>
131 effect.suspend(() => {
132 let i = 0
133 let meta: Api.Metadata = {
134 attempt: 0,
135 stepIndex: 0
136 }
137 const provideMeta = effect.provideServiceEffect(
138 Api.CurrentMetadata,
139 effect.sync(() => {
140 meta = {
141 attempt: meta.attempt + 1,
142 stepIndex: i
143 }
144 return meta
145 })
146 )
147 const emitter = options?.onEvent === undefined
148 ? undefined
149 : makeEventEmitter(options.onEvent, () => meta)
150 const instrument: (attempt: Effect<A, any, any>) => Effect<A, any, any> = emitter === undefined
151 ? identity
152 : (attempt) =>
153 effect.uninterruptibleMask((restore) =>
154 effect.flatMap(emitter.begin, (state) => effect.onExit(restore(attempt), (exit) => emitter.end(state, exit)))
155 )
156 let result: Result.Result<A, any> | undefined
157 return effect.flatMap(
158 effect.whileLoop({
159 while: () => i < plan.steps.length && (result === undefined || Result.isFailure(result)),
160 body() {
161 const step = plan.steps[i]
162 let nextEffect: Effect<A, any, any> = provideMeta(
163 instrument(internalLayer.provide(self, step.provide as any))
164 )
165 if (result) {
166 let attempted = false
167 const wrapped = nextEffect
168 // ensure the schedule is applied at least once
169 nextEffect = effect.suspend(() => {
170 if (attempted) return wrapped
171 attempted = true
172 return effect.fromResult(result!)
173 })
174 nextEffect = internalSchedule.retry(nextEffect, scheduleFromStep(step, false)!)
175 } else {
176 const schedule = scheduleFromStep(step, true)
177 nextEffect = schedule ? internalSchedule.retry(nextEffect, schedule) : nextEffect
178 }
179 return effect.result(nextEffect)
180 },
181 step(result_) {
182 result = result_
183 i++
184 }
185 }),
186 () => effect.fromResult(result!)
187 )
188 }))
189
190/** @internal */
191export const scheduleFromStep = <Provides, In, PlanE, PlanR>(
192 step: Api.ExecutionPlan<{
193 provides: Provides
194 input: In
195 error: PlanE
196 requirements: PlanR
197 }>["steps"][number],
198 first: boolean
199) => {
200 if (!first) {
201 return internalSchedule.buildFromOptions({
202 schedule: step.schedule ? step.schedule : step.attempts ? undefined : scheduleOnce,
203 times: step.attempts,
204 while: step.while
205 })
206 } else if (step.attempts === 1 || !(step.schedule || step.attempts)) {
207 return undefined
208 }
209 return internalSchedule.buildFromOptions({
210 schedule: step.schedule,
211 while: step.while,
212 times: step.attempts ? step.attempts - 1 : undefined
213 })
214}
215
216const scheduleOnce = Schedule.recurs(1)