1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Effect, Exit, Layer, Option, Schema } from "effect"
3import { PersistedQueue } from "effect/persistence"
4import { TestClock } from "effect/testing"
5import { DurableQueue, Workflow, WorkflowEngine } from "effect/workflow"
6
7const PersistedQueueLayer = PersistedQueue.layer.pipe(
8 Layer.provideMerge(PersistedQueue.layerStoreMemory)
9)
10
11const pollUntilComplete = <A, E, R>(
12 poll: Effect.Effect<Option.Option<Workflow.Result<A, E>>, never, R>
13) =>
14 Effect.gen(function*() {
15 let polled = yield* poll
16 for (let i = 0; i < 10 && (Option.isNone(polled) || polled.value._tag !== "Complete"); i++) {
17 yield* Effect.yieldNow
18 yield* Effect.sleep("10 millis").pipe(TestClock.withLive)
19 polled = yield* poll
20 }
21 return polled
22 })
23
24describe("DurableQueue", () => {
25 const SuccessQueue = DurableQueue.make({
26 name: "DurableQueueTest/SuccessQueue",
27 payload: {
28 id: Schema.String,
29 value: Schema.Number
30 },
31 success: Schema.Number,
32 error: Schema.String,
33 idempotencyKey: ({ id }) => id
34 })
35
36 const SuccessWorkflow = Workflow.make("DurableQueueTest/SuccessWorkflow", {
37 payload: {
38 id: Schema.String,
39 value: Schema.Number
40 },
41 success: Schema.Number,
42 error: Schema.String,
43 idempotencyKey: ({ id }) => id
44 })
45
46 const SuccessLayer = Layer.mergeAll(
47 SuccessWorkflow.toLayer(({ id, value }) => DurableQueue.process(SuccessQueue, { id, value })),
48 DurableQueue.worker(
49 SuccessQueue,
50 ({ value }) => Effect.succeed(value + 1)
51 )
52 ).pipe(
53 Layer.provideMerge(WorkflowEngine.layerMemory),
54 Layer.provideMerge(PersistedQueueLayer)
55 )
56
57 it.effect("processes queued items and resumes the workflow", () =>
58 Effect.gen(function*() {
59 const executionId = yield* SuccessWorkflow.execute({ id: "success", value: 41 }, { discard: true })
60 const polled = yield* pollUntilComplete(SuccessWorkflow.poll(executionId))
61
62 assert(Option.isSome(polled) && polled.value._tag === "Complete" && Exit.isSuccess(polled.value.exit))
63 assert.strictEqual(polled.value.exit.value, 42)
64 }).pipe(Effect.provide(SuccessLayer)))
65
66 const FailureQueue = DurableQueue.make({
67 name: "DurableQueueTest/FailureQueue",
68 payload: {
69 id: Schema.String
70 },
71 success: Schema.Void,
72 error: Schema.String,
73 idempotencyKey: ({ id }) => id
74 })
75
76 const FailureWorkflow = Workflow.make("DurableQueueTest/FailureWorkflow", {
77 payload: {
78 id: Schema.String
79 },
80 success: Schema.Void,
81 error: Schema.String,
82 idempotencyKey: ({ id }) => id
83 })
84
85 const FailureLayer = Layer.mergeAll(
86 FailureWorkflow.toLayer(({ id }) => DurableQueue.process(FailureQueue, { id })),
87 DurableQueue.worker(
88 FailureQueue,
89 () => Effect.fail("boom")
90 )
91 ).pipe(
92 Layer.provideMerge(WorkflowEngine.layerMemory),
93 Layer.provideMerge(PersistedQueueLayer)
94 )
95
96 it.effect("propagates worker failures to the workflow", () =>
97 Effect.gen(function*() {
98 const executionId = yield* FailureWorkflow.execute({ id: "failure" }, { discard: true })
99 const polled = yield* pollUntilComplete(FailureWorkflow.poll(executionId))
100
101 assert(Option.isSome(polled) && polled.value._tag === "Complete" && Exit.isFailure(polled.value.exit))
102 const failure = polled.value.exit.cause.reasons.find(Cause.isFailReason)
103 assert.strictEqual(failure?.error, "boom")
104 }).pipe(Effect.provide(FailureLayer)))
105})