106 lines
3.4 KiB
TypeScript
106 lines
3.4 KiB
TypeScript
import { assert, describe, it } from "@effect/vitest"
|
|
import { Cause, Effect, Exit, Layer, Option, Schema } from "effect"
|
|
import { TestClock } from "effect/testing"
|
|
import { PersistedQueue } from "effect/unstable/persistence"
|
|
import { DurableQueue, Workflow, WorkflowEngine } from "effect/unstable/workflow"
|
|
|
|
const PersistedQueueLayer = PersistedQueue.layer.pipe(
|
|
Layer.provideMerge(PersistedQueue.layerStoreMemory)
|
|
)
|
|
|
|
const pollUntilComplete = <A, E, R>(
|
|
poll: Effect.Effect<Option.Option<Workflow.Result<A, E>>, never, R>
|
|
) =>
|
|
Effect.gen(function*() {
|
|
let polled = yield* poll
|
|
for (let i = 0; i < 10 && (Option.isNone(polled) || polled.value._tag !== "Complete"); i++) {
|
|
yield* Effect.yieldNow
|
|
yield* Effect.sleep("10 millis").pipe(TestClock.withLive)
|
|
polled = yield* poll
|
|
}
|
|
return polled
|
|
})
|
|
|
|
describe("DurableQueue", () => {
|
|
const SuccessQueue = DurableQueue.make({
|
|
name: "DurableQueueTest/SuccessQueue",
|
|
payload: {
|
|
id: Schema.String,
|
|
value: Schema.Number
|
|
},
|
|
success: Schema.Number,
|
|
error: Schema.String,
|
|
idempotencyKey: ({ id }) => id
|
|
})
|
|
|
|
const SuccessWorkflow = Workflow.make("DurableQueueTest/SuccessWorkflow", {
|
|
payload: {
|
|
id: Schema.String,
|
|
value: Schema.Number
|
|
},
|
|
success: Schema.Number,
|
|
error: Schema.String,
|
|
idempotencyKey: ({ id }) => id
|
|
})
|
|
|
|
const SuccessLayer = Layer.mergeAll(
|
|
SuccessWorkflow.toLayer(({ id, value }) => DurableQueue.process(SuccessQueue, { id, value })),
|
|
DurableQueue.worker(
|
|
SuccessQueue,
|
|
({ value }) => Effect.succeed(value + 1)
|
|
)
|
|
).pipe(
|
|
Layer.provideMerge(WorkflowEngine.layerMemory),
|
|
Layer.provideMerge(PersistedQueueLayer)
|
|
)
|
|
|
|
it.effect("processes queued items and resumes the workflow", () =>
|
|
Effect.gen(function*() {
|
|
const executionId = yield* SuccessWorkflow.execute({ id: "success", value: 41 }, { discard: true })
|
|
const polled = yield* pollUntilComplete(SuccessWorkflow.poll(executionId))
|
|
|
|
assert(Option.isSome(polled) && polled.value._tag === "Complete" && Exit.isSuccess(polled.value.exit))
|
|
assert.strictEqual(polled.value.exit.value, 42)
|
|
}).pipe(Effect.provide(SuccessLayer)))
|
|
|
|
const FailureQueue = DurableQueue.make({
|
|
name: "DurableQueueTest/FailureQueue",
|
|
payload: {
|
|
id: Schema.String
|
|
},
|
|
success: Schema.Void,
|
|
error: Schema.String,
|
|
idempotencyKey: ({ id }) => id
|
|
})
|
|
|
|
const FailureWorkflow = Workflow.make("DurableQueueTest/FailureWorkflow", {
|
|
payload: {
|
|
id: Schema.String
|
|
},
|
|
success: Schema.Void,
|
|
error: Schema.String,
|
|
idempotencyKey: ({ id }) => id
|
|
})
|
|
|
|
const FailureLayer = Layer.mergeAll(
|
|
FailureWorkflow.toLayer(({ id }) => DurableQueue.process(FailureQueue, { id })),
|
|
DurableQueue.worker(
|
|
FailureQueue,
|
|
() => Effect.fail("boom")
|
|
)
|
|
).pipe(
|
|
Layer.provideMerge(WorkflowEngine.layerMemory),
|
|
Layer.provideMerge(PersistedQueueLayer)
|
|
)
|
|
|
|
it.effect("propagates worker failures to the workflow", () =>
|
|
Effect.gen(function*() {
|
|
const executionId = yield* FailureWorkflow.execute({ id: "failure" }, { discard: true })
|
|
const polled = yield* pollUntilComplete(FailureWorkflow.poll(executionId))
|
|
|
|
assert(Option.isSome(polled) && polled.value._tag === "Complete" && Exit.isFailure(polled.value.exit))
|
|
const failure = polled.value.exit.cause.reasons.find(Cause.isFailReason)
|
|
assert.strictEqual(failure?.error, "boom")
|
|
}).pipe(Effect.provide(FailureLayer)))
|
|
})
|