1/**
2 * Durable workflow queues delegate work to persisted background workers and
3 * resume the waiting workflow with the worker result.
4 *
5 * A workflow calls `process` to encode a payload, offer it to a named
6 * `PersistedQueue`, attach a `DurableDeferred` token, and suspend. A worker
7 * created with `makeWorker` or `worker` takes the item, runs the handler, and
8 * records the handler's `Exit` through that token so the original workflow can
9 * continue with the typed success or error.
10 *
11 * Delivery is at-least-once: a crash between handler success and the
12 * acknowledgement redelivers the item, so handlers must be idempotent. When an
13 * item exhausts its persisted queue attempts it is dead-lettered: the
14 * `DurableDeferred` never resolves and the workflow stays parked until the
15 * item is requeued out of band, while the id-based de-duplication prevents
16 * replays from resurrecting the failed item. Deployments should run
17 * `PersistedQueue.layerCleanup` in one instance to prune old completed items.
18 *
19 * @stability unstable
20 * @since 4.0.0
21 */
22import * as Effect from "../Effect.ts"
23import * as Layer from "../Layer.ts"
24import * as PersistedQueue from "../persistence/PersistedQueue.ts"
25import * as Schedule from "../Schedule.ts"
26import * as Schema from "../Schema.ts"
27import * as Tracer from "../Tracer.ts"
28import * as Activity from "./Activity.ts"
29import * as DurableDeferred from "./DurableDeferred.ts"
30import type { WorkflowEngine, WorkflowInstance } from "./WorkflowEngine.ts"
31
32/**
33 * Type-level identifier used to recognize `DurableQueue` values.
34 *
35 * @stability unstable
36 * @category type IDs
37 * @since 4.0.0
38 */
39export type TypeId = "~effect/workflow/DurableQueue"
40
41/**
42 * Runtime identifier attached to `DurableQueue` values.
43 *
44 * @stability unstable
45 * @category type IDs
46 * @since 4.0.0
47 */
48export const TypeId: TypeId = "~effect/workflow/DurableQueue"
49
50/**
51 * Durable workflow queue definition containing a payload schema, idempotency
52 * key, and deferred used to await worker results.
53 *
54 * @stability unstable
55 * @category models
56 * @since 4.0.0
57 */
58export interface DurableQueue<
59 Payload extends Schema.Top,
60 Success extends Schema.Top = Schema.Void,
61 Error extends Schema.Top = Schema.Never
62> {
63 readonly [TypeId]: TypeId
64 readonly name: string
65 readonly payloadSchema: Payload
66 readonly idempotencyKey: (payload: Payload["Type"]) => string
67 readonly deferred: DurableDeferred.DurableDeferred<Success, Error>
68}
69
70/**
71 * Creates a `DurableQueue` that waits for persisted items to finish processing
72 * using a `DurableDeferred`.
73 *
74 * **Example** (Defining a durable queue with workers)
75 *
76 * ```ts import.meta.vitest
77 * import { Effect, Layer, Schema } from "effect"
78 * import { DurableQueue, Workflow } from "effect/workflow"
79 *
80 * // Define a DurableQueue that can be used to derive workers and offer items for
81 * // processing.
82 * const ApiQueue = DurableQueue.make({
83 * name: "ApiQueue",
84 * payload: {
85 * id: Schema.String
86 * },
87 * success: Schema.Void,
88 * error: Schema.Never,
89 * idempotencyKey(payload) {
90 * return payload.id
91 * }
92 * })
93 *
94 * const MyWorkflow = Workflow.make("MyWorkflow", {
95 * payload: {
96 * id: Schema.String
97 * },
98 * idempotencyKey: ({ id }) => id
99 * })
100 *
101 * const MyWorkflowLayer = MyWorkflow.toLayer(
102 * Effect.fnUntraced(function*() {
103 * // The workflow suspends until a worker completes this queue item.
104 * yield* DurableQueue.process(ApiQueue, { id: "api-call-1" })
105 * return "Workflow succeeded!"
106 * })
107 * )
108 *
109 * const processed: Array<string> = []
110 * const processApiCall = ({ id }: { readonly id: string }) => Effect.sync(() => processed.push(id))
111 *
112 * // Construct the worker layer without starting background workers in this example.
113 * const ApiWorker = DurableQueue.worker(ApiQueue, processApiCall, {
114 * concurrency: 5
115 * })
116 *
117 * const program = Effect.gen(function*() {
118 * // Exercise the finite handler directly instead of running a queue worker.
119 * yield* processApiCall({ id: "api-call-1" })
120 * return [Layer.isLayer(MyWorkflowLayer), Layer.isLayer(ApiWorker), processed] as const
121 * })
122 *
123 * await Effect.runPromise(program) // => [true, true, ["api-call-1"]]
124 * ```
125 *
126 * @stability unstable
127 * @category constructors
128 * @since 4.0.0
129 */
130export const make = <
131 Payload extends Schema.Top | Schema.Struct.Fields,
132 Success extends Schema.Top = Schema.Void,
133 Error extends Schema.Top = Schema.Never
134>(options: {
135 readonly name: string
136 readonly payload: Payload
137 readonly idempotencyKey: (
138 payload: Payload extends Schema.Struct.Fields ? Schema.Struct.Type<Payload>
139 : Payload["Type"]
140 ) => string
141 readonly success?: Success | undefined
142 readonly error?: Error | undefined
143}): DurableQueue<
144 Payload extends Schema.Struct.Fields ? Schema.Struct<Payload> : Payload,
145 Success,
146 Error
147> => ({
148 [TypeId]: TypeId,
149 name: options.name,
150 payloadSchema: Schema.isSchema(options.payload)
151 ? options.payload
152 : Schema.Struct(options.payload as any) as any,
153 idempotencyKey: options.idempotencyKey as any,
154 deferred: DurableDeferred.make(`DurableQueue/${options.name}`, {
155 success: options.success,
156 error: options.error
157 })
158})
159
160const queueSchemas = new WeakMap<Schema.Top, Schema.Top>()
161
162const getQueueSchema = <Payload extends Schema.Top>(
163 payload: Payload
164): Schema.Struct<{
165 token: typeof DurableDeferred.Token
166 payload: Payload
167 traceId: typeof Schema.String
168 spanId: typeof Schema.String
169 sampled: typeof Schema.Boolean
170}> => {
171 let schema = queueSchemas.get(payload)
172 if (!schema) {
173 schema = Schema.Struct({
174 token: DurableDeferred.Token,
175 traceId: Schema.String,
176 spanId: Schema.String,
177 sampled: Schema.Boolean,
178 payload
179 })
180 queueSchemas.set(payload, schema)
181 }
182 return schema as any
183}
184
185/**
186 * Adds an item to the queue and wait for a worker to process it.
187 *
188 * @stability unstable
189 * @category running
190 * @since 4.0.0
191 */
192export const process: <
193 Payload extends Schema.Top,
194 Success extends Schema.Top,
195 Error extends Schema.Top
196>(
197 self: DurableQueue<Payload, Success, Error>,
198 payload: Payload["~type.make.in"],
199 options?: {
200 readonly retrySchedule?: Schedule.Schedule<any, PersistedQueue.PersistedQueueError> | undefined
201 }
202) => Effect.Effect<
203 Success["Type"],
204 Error["Type"],
205 | WorkflowEngine
206 | WorkflowInstance
207 | PersistedQueue.PersistedQueueFactory
208 | Payload["EncodingServices"]
209 | Payload["DecodingServices"]
210 | Success["DecodingServices"]
211 | Error["DecodingServices"]
212> = Effect.fnUntraced(function*<
213 Payload extends Schema.Top,
214 Success extends Schema.Top,
215 Error extends Schema.Top
216>(self: DurableQueue<Payload, Success, Error>, fields: Payload["~type.make.in"], options?: {
217 readonly retrySchedule?: Schedule.Schedule<any, PersistedQueue.PersistedQueueError> | undefined
218}) {
219 const payload = self.payloadSchema.make(fields)
220 const queueName = `DurableQueue/${self.name}`
221 const queue = yield* PersistedQueue.make({
222 name: queueName,
223 schema: getQueueSchema(self.payloadSchema)
224 })
225 const id = yield* Activity.idempotencyKey(`${queueName}/${self.idempotencyKey(payload)}`)
226
227 const deferred = DurableDeferred.make(`${self.deferred.name}/${id}`, {
228 success: self.deferred.successSchema,
229 error: self.deferred.errorSchema
230 })
231 const token = yield* DurableDeferred.token(deferred)
232
233 yield* Effect.useSpan(`DurableQueue/${self.name}/process`, {
234 attributes: { id }
235 }, (span) =>
236 queue.offer({
237 token,
238 payload,
239 traceId: span.traceId,
240 spanId: span.spanId,
241 sampled: span.sampled
242 } as any, { id }).pipe(
243 Effect.tapCause(Effect.logWarning),
244 Effect.catchTag("SchemaError", Effect.die),
245 Effect.retry(options?.retrySchedule ?? defaultRetrySchedule),
246 Effect.orDie,
247 Effect.annotateLogs({
248 package: "effect",
249 module: "DurableQueue",
250 fiber: "process",
251 queueName: self.name
252 })
253 ))
254
255 return yield* DurableDeferred.await(deferred)
256})
257
258const defaultRetrySchedule = Schedule.min([
259 Schedule.exponential(500, 1.5),
260 Schedule.spaced("1 minute")
261])
262
263/**
264 * Create a worker effect that processes items from the durable queue.
265 *
266 * @stability unstable
267 * @category workers
268 * @since 4.0.0
269 */
270export const makeWorker: <
271 Payload extends Schema.Top,
272 Success extends Schema.Top,
273 Error extends Schema.Top,
274 R
275>(
276 self: DurableQueue<Payload, Success, Error>,
277 f: (payload: Payload["Type"]) => Effect.Effect<Success["Type"], Error["Type"], R>,
278 options?: { readonly concurrency?: number | undefined } | undefined
279) => Effect.Effect<
280 never,
281 never,
282 | WorkflowEngine
283 | PersistedQueue.PersistedQueueFactory
284 | R
285 | Payload["EncodingServices"]
286 | Payload["DecodingServices"]
287 | Success["EncodingServices"]
288 | Error["EncodingServices"]
289> = Effect.fnUntraced(function*<
290 Payload extends Schema.Top,
291 Success extends Schema.Top,
292 Error extends Schema.Top,
293 R
294>(
295 self: DurableQueue<Payload, Success, Error>,
296 f: (payload: Payload["Type"]) => Effect.Effect<Success["Type"], Error["Type"], R>,
297 options?: {
298 readonly concurrency?: number | undefined
299 }
300) {
301 const queue = yield* PersistedQueue.make({
302 name: `DurableQueue/${self.name}`,
303 schema: getQueueSchema(self.payloadSchema)
304 })
305 const concurrency = options?.concurrency ?? 1
306
307 const worker = queue.take((item_) => {
308 const item = item_ as {
309 readonly token: DurableDeferred.Token
310 readonly payload: Payload["Type"]
311 readonly traceId: string
312 readonly spanId: string
313 readonly sampled: boolean
314 }
315 return Effect.withSpan(
316 f(item.payload).pipe(
317 Effect.exit,
318 Effect.flatMap((exit) =>
319 DurableDeferred.done(self.deferred, {
320 token: item.token,
321 exit
322 })
323 ),
324 Effect.asVoid
325 ),
326 `DurableQueue/${self.name}/worker`,
327 {
328 captureStackTrace: false,
329 parent: Tracer.externalSpan({
330 traceId: item.traceId,
331 spanId: item.spanId,
332 sampled: item.sampled
333 })
334 }
335 )
336 }).pipe(
337 Effect.catchCause(Effect.logWarning),
338 Effect.forever,
339 Effect.annotateLogs({
340 package: "effect",
341 module: "DurableQueue",
342 fiber: "worker",
343 queueName: self.name
344 })
345 )
346
347 yield* Effect.replicateEffect(worker, concurrency, { concurrency, discard: true })
348 return yield* Effect.never
349})
350
351/**
352 * Create a layer that runs workers for the durable queue.
353 *
354 * @stability unstable
355 * @category workers
356 * @since 4.0.0
357 */
358export const worker: <
359 Payload extends Schema.Top,
360 Success extends Schema.Top,
361 Error extends Schema.Top,
362 R
363>(
364 self: DurableQueue<Payload, Success, Error>,
365 f: (payload: Payload["Type"]) => Effect.Effect<Success["Type"], Error["Type"], R>,
366 options?: {
367 readonly concurrency?: number | undefined
368 } | undefined
369) => Layer.Layer<
370 never,
371 never,
372 | WorkflowEngine
373 | PersistedQueue.PersistedQueueFactory
374 | R
375 | Payload["EncodingServices"]
376 | Payload["DecodingServices"]
377 | Success["EncodingServices"]
378 | Error["EncodingServices"]
379> = (self, f, options) => Layer.effectDiscard(Effect.forkScoped(makeWorker(self, f, options)))