51 lines
1.4 KiB
TypeScript
51 lines
1.4 KiB
TypeScript
import { expect, test } from "bun:test";
|
|
import { createDb } from "@wrnexus/db";
|
|
import { sqlite } from "@wrnexus/db/sqlite";
|
|
import {
|
|
createDurableQueue,
|
|
defineSaga,
|
|
drainOutbox,
|
|
memoryQueueStore,
|
|
testQueue,
|
|
withOutbox,
|
|
} from "../src/index.ts";
|
|
|
|
test("outbox commits intents and deterministic queue helpers dispatch them", async () => {
|
|
const db = createDb(sqlite(":memory:"));
|
|
const queue = createDurableQueue({ store: memoryQueueStore() });
|
|
const seen: number[] = [];
|
|
queue.process("email:send", async (job) => void seen.push((job.data as { id: number }).id));
|
|
await withOutbox(db, async ({ enqueue }) => enqueue({ name: "email:send" } as never, { id: 7 }));
|
|
expect(
|
|
await drainOutbox(db, (name) =>
|
|
name === "email:send"
|
|
? { add: (data, options) => queue.add(name, data, options) }
|
|
: undefined,
|
|
),
|
|
).toBe(1);
|
|
expect(await testQueue(queue).runAll()).toBe(1);
|
|
expect(seen).toEqual([7]);
|
|
});
|
|
|
|
test("sagas compensate completed steps in reverse order", async () => {
|
|
const calls: string[] = [];
|
|
const saga = defineSaga({
|
|
name: "send",
|
|
steps: [
|
|
{
|
|
name: "reserve",
|
|
run: () => void calls.push("reserve"),
|
|
compensate: () => void calls.push("refund"),
|
|
},
|
|
{
|
|
name: "queue",
|
|
run: () => {
|
|
throw new Error("down");
|
|
},
|
|
},
|
|
],
|
|
});
|
|
await expect(saga.run({})).rejects.toThrow("down");
|
|
expect(calls).toEqual(["reserve", "refund"]);
|
|
});
|