mirror of
https://github.com/devkitshq/notifkit.git
synced 2026-10-19 18:24:52 +02:00
659 lines
20 KiB
TypeScript
659 lines
20 KiB
TypeScript
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
|
import { WorkflowWorker } from "@/services/workflow/main.js";
|
|
import { workflowRegistry } from "@/workflows/index.js";
|
|
import type { StreamMessage } from "@/queue/index.js";
|
|
import { buildStreamEvent } from "@/contracts/index.js";
|
|
|
|
describe("WorkflowWorker Edge Cases", () => {
|
|
let mockRedis: any;
|
|
let mockDb: any;
|
|
let mockWorkflowProducer: any;
|
|
let mockNotificationProducer: any;
|
|
let worker: WorkflowWorker;
|
|
|
|
beforeEach(() => {
|
|
mockRedis = {
|
|
set: vi.fn().mockResolvedValue("OK"),
|
|
releaseLock: vi.fn().mockResolvedValue(1),
|
|
renewLock: vi.fn().mockResolvedValue(1),
|
|
zadd: vi.fn().mockResolvedValue(1),
|
|
};
|
|
|
|
const qb: any = {};
|
|
qb.select = vi.fn().mockReturnValue(qb);
|
|
qb.from = vi.fn().mockReturnValue(qb);
|
|
qb.where = vi.fn().mockReturnValue(qb);
|
|
qb.limit = vi
|
|
.fn()
|
|
.mockResolvedValue([{ id: "inst-1", status: "pending", name: "test_flow", input: {} }]);
|
|
qb.insert = vi.fn().mockReturnValue(qb);
|
|
qb.values = vi.fn().mockReturnValue(qb);
|
|
qb.returning = vi
|
|
.fn()
|
|
.mockResolvedValue([{ id: "inst-1", status: "pending", name: "test_flow", input: {} }]);
|
|
qb.update = vi.fn().mockReturnValue(qb);
|
|
qb.set = vi.fn().mockReturnValue(qb);
|
|
qb.then = function (resolve: any) {
|
|
resolve([]);
|
|
};
|
|
|
|
mockDb = qb;
|
|
|
|
mockWorkflowProducer = {
|
|
publishBatch: vi.fn().mockResolvedValue({ messageIds: ["1", "2"], eventIds: ["a", "b"] }),
|
|
};
|
|
mockNotificationProducer = {
|
|
publishBatch: vi.fn().mockResolvedValue({ messageIds: ["1", "2"], eventIds: ["a", "b"] }),
|
|
};
|
|
|
|
worker = new WorkflowWorker({
|
|
consumer: { ack: vi.fn(), nack: vi.fn() } as any,
|
|
pendingScanner: {} as any,
|
|
logger: {
|
|
child: vi.fn().mockReturnThis(),
|
|
info: vi.fn(),
|
|
warn: vi.fn(),
|
|
error: vi.fn(),
|
|
debug: vi.fn(),
|
|
} as any,
|
|
concurrency: 1,
|
|
redis: mockRedis,
|
|
db: mockDb,
|
|
workflowProducer: mockWorkflowProducer,
|
|
notificationProducer: mockNotificationProducer,
|
|
});
|
|
});
|
|
|
|
afterEach(async () => {
|
|
vi.clearAllMocks();
|
|
await worker.stop();
|
|
});
|
|
|
|
it("should ignore events that are not workflow.triggered or resumed", async () => {
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent("notification.created" as any, {}, "test", "trace") as any,
|
|
};
|
|
await worker.process(msg);
|
|
expect(mockRedis.set).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("should skip if workflow instance is missing name, instanceId, or projectId", async () => {
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent("workflow.triggered", { name: "test_flow" }, "test", "trace") as any, // missing instanceId/projectId
|
|
};
|
|
await worker.process(msg);
|
|
expect(mockRedis.set).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("should skip if no handler found for workflow", async () => {
|
|
mockDb.limit.mockResolvedValueOnce([]); // No dynamic definition found either
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "missing_flow", instanceId: "1", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
await worker.process(msg);
|
|
expect(mockRedis.set).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("should skip if lock cannot be acquired", async () => {
|
|
workflowRegistry.register("test_flow", async () => {});
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_flow", instanceId: "inst-1", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
mockRedis.set.mockResolvedValue(null); // lock acquisition fails
|
|
await worker.process(msg);
|
|
expect(mockDb.select).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("should handle wait step with correct parsing (seconds)", async () => {
|
|
let executed = false;
|
|
workflowRegistry.register("test_flow", async (ctx) => {
|
|
executed = true;
|
|
await ctx.step.wait("5s");
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_flow", instanceId: "inst-1", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([{ id: "inst-1", status: "pending", name: "test_flow", input: {} }])
|
|
.mockResolvedValueOnce([]); // No existing steps
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(executed).toBe(true);
|
|
expect(mockRedis.zadd).toHaveBeenCalled();
|
|
const args = mockRedis.zadd.mock.calls[0];
|
|
expect(args[0]).toBe("notif:workflow:timers");
|
|
|
|
// DB update to pending should be called because of SuspendExecutionError
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "pending" });
|
|
});
|
|
|
|
it("should handle run step", async () => {
|
|
let executedResult = null;
|
|
workflowRegistry.register("test_flow", async (ctx) => {
|
|
executedResult = await ctx.step.run("custom_logic", async () => {
|
|
return { custom: "result" };
|
|
});
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_flow", instanceId: "inst-1", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([{ id: "inst-1", status: "pending", name: "test_flow", input: {} }])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
expect(executedResult).toEqual({ custom: "result" });
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "completed" });
|
|
});
|
|
|
|
it("should fail workflow if handler throws unhandled error", async () => {
|
|
workflowRegistry.register("test_flow", async () => {
|
|
throw new Error("Unhandled crash");
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_flow", instanceId: "inst-1", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit.mockResolvedValueOnce([
|
|
{ id: "inst-1", status: "pending", name: "test_flow", input: {} },
|
|
]);
|
|
|
|
await worker.process(msg);
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "failed" });
|
|
});
|
|
|
|
it("should enforce 24h default timeout for waitForEvent if none provided", async () => {
|
|
let executed = false;
|
|
workflowRegistry.register("test_flow", async (ctx) => {
|
|
executed = true;
|
|
await ctx.step.waitForEvent("payment.succeeded");
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_flow", instanceId: "inst-1", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([{ id: "inst-1", status: "pending", name: "test_flow", input: {} }])
|
|
.mockResolvedValueOnce([]); // No existing steps
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(executed).toBe(true);
|
|
expect(mockRedis.zadd).toHaveBeenCalled();
|
|
const args = mockRedis.zadd.mock.calls[0];
|
|
expect(args[0]).toBe("notif:workflow:timers");
|
|
// Ensure timeout is approx Date.now() + 24 hours (86400000 ms)
|
|
const score = args[1];
|
|
const expectedScore = Date.now() + 86400000;
|
|
expect(score).toBeGreaterThan(expectedScore - 5000);
|
|
expect(score).toBeLessThan(expectedScore + 5000);
|
|
});
|
|
|
|
it("gracefully handles bad template ID in notify step", async () => {
|
|
let executed = false;
|
|
workflowRegistry.register("test_flow", async (ctx) => {
|
|
executed = true;
|
|
await ctx.step.notify({
|
|
template: "missing-template",
|
|
user: "u-1",
|
|
channels: ["email"],
|
|
});
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_flow", instanceId: "inst-1", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([{ id: "inst-1", status: "pending", name: "test_flow", input: {} }])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(executed).toBe(true);
|
|
// Should publish the request but not crash the workflow
|
|
expect(mockNotificationProducer.publishBatch).toHaveBeenCalledTimes(1);
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "completed" });
|
|
});
|
|
|
|
it("handles days unit ('7d') in waitForEvent timeout correctly", async () => {
|
|
let executed = false;
|
|
workflowRegistry.register("test_days_flow", async (ctx) => {
|
|
executed = true;
|
|
await ctx.step.waitForEvent("user.login", { timeout: "7d" });
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_days_flow", instanceId: "inst-7d", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-7d", status: "pending", name: "test_days_flow", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(executed).toBe(true);
|
|
expect(mockRedis.zadd).toHaveBeenCalled();
|
|
const args = mockRedis.zadd.mock.calls[0];
|
|
const score = args[1];
|
|
const expectedScore = Date.now() + 7 * 24 * 60 * 60 * 1000;
|
|
expect(score).toBeGreaterThan(expectedScore - 5000);
|
|
expect(score).toBeLessThan(expectedScore + 5000);
|
|
});
|
|
|
|
it("fails workflow on invalid waitForEvent timeout duration", async () => {
|
|
workflowRegistry.register("test_bad_timeout", async (ctx) => {
|
|
await ctx.step.waitForEvent("user.login", { timeout: "invalid_duration" as any });
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_bad_timeout", instanceId: "inst-bad", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-bad", status: "pending", name: "test_bad_timeout", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "failed" });
|
|
});
|
|
|
|
it("handles dynamic JSON workflow execution from database definitions", async () => {
|
|
// No code-registered handler; fetch from db
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "dynamic_json_flow", instanceId: "inst-dyn", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
// 1. Definition lookup
|
|
.mockResolvedValueOnce([
|
|
{
|
|
projectId: "p1",
|
|
name: "dynamic_json_flow",
|
|
steps: [
|
|
{ action: "notify", payload: { user: "u-1", channels: ["email"], template: "tpl-1" } },
|
|
{ action: "wait", duration: "10s" },
|
|
],
|
|
},
|
|
])
|
|
// 2. Instance lookup
|
|
.mockResolvedValueOnce([
|
|
{
|
|
id: "inst-dyn",
|
|
status: "pending",
|
|
name: "dynamic_json_flow",
|
|
input: { user: { id: "u-1" } },
|
|
},
|
|
])
|
|
// Then the per-step cancellation checks, which find nothing canceled.
|
|
.mockResolvedValue([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(mockNotificationProducer.publishBatch).toHaveBeenCalledTimes(1);
|
|
expect(mockRedis.zadd).toHaveBeenCalled();
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "pending" });
|
|
});
|
|
|
|
it("handles concurrent resume collision gracefully when lock is already held", async () => {
|
|
workflowRegistry.register("test_lock_race", async (ctx) => {
|
|
await ctx.step.wait("10s");
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.resumed",
|
|
{ name: "test_lock_race", instanceId: "inst-locked", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
// Lock acquisition returns null (another process is currently executing)
|
|
mockRedis.set.mockResolvedValueOnce(null);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(mockDb.update).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("returns null from step.waitForEvent when step output has timedOut: true", async () => {
|
|
let returnedEvent: any = "unset";
|
|
workflowRegistry.register("test_timeout_replay", async (ctx) => {
|
|
returnedEvent = await ctx.step.waitForEvent("payment.received", { timeout: "1h" });
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.resumed",
|
|
{ name: "test_timeout_replay", instanceId: "inst-timeout-replay", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-timeout-replay", status: "pending", name: "test_timeout_replay", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
// Provide existing step with output = { timedOut: true }
|
|
mockDb.then = function (resolve: any) {
|
|
resolve([
|
|
{ id: "step-1", stepIndex: "0", action: "waitForEvent", output: { timedOut: true } },
|
|
]);
|
|
};
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(returnedEvent).toBeNull();
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "completed" });
|
|
});
|
|
|
|
it("handles minutes ('15m') and hours ('3h') units in step.wait correctly", async () => {
|
|
workflowRegistry.register("test_units_flow", async (ctx) => {
|
|
await ctx.step.wait("15m");
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_units_flow", instanceId: "inst-units", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-units", status: "pending", name: "test_units_flow", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(mockRedis.zadd).toHaveBeenCalled();
|
|
const score = mockRedis.zadd.mock.calls[0]![1];
|
|
const expectedScore = Date.now() + 15 * 60 * 1000;
|
|
expect(score).toBeGreaterThan(expectedScore - 5000);
|
|
expect(score).toBeLessThan(expectedScore + 5000);
|
|
});
|
|
|
|
it("fails workflow on invalid step.wait duration format", async () => {
|
|
workflowRegistry.register("test_bad_wait", async (ctx) => {
|
|
await ctx.step.wait("100x" as any);
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_bad_wait", instanceId: "inst-bad-wait", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-bad-wait", status: "pending", name: "test_bad_wait", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "failed" });
|
|
});
|
|
|
|
it("skips step.run execution on replay when step output is already memoized", async () => {
|
|
let runCallCount = 0;
|
|
let finalResult = null;
|
|
|
|
workflowRegistry.register("test_memoized_run", async (ctx) => {
|
|
finalResult = await ctx.step.run("calculate_val", async () => {
|
|
runCallCount++;
|
|
return { calculated: 42 };
|
|
});
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.resumed",
|
|
{ name: "test_memoized_run", instanceId: "inst-memo-run", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-memo-run", status: "pending", name: "test_memoized_run", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
// Provide existing step with output = { calculated: 42 }
|
|
mockDb.then = function (resolve: any) {
|
|
resolve([{ id: "step-1", stepIndex: "0", action: "run", output: { calculated: 42 } }]);
|
|
};
|
|
|
|
await worker.process(msg);
|
|
|
|
// fn() should never have been invoked because it was memoized!
|
|
expect(runCallCount).toBe(0);
|
|
expect(finalResult).toEqual({ calculated: 42 });
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "completed" });
|
|
});
|
|
|
|
it("fails workflow if step.run callback throws an error", async () => {
|
|
workflowRegistry.register("test_throw_run", async (ctx) => {
|
|
await ctx.step.run("failing_step", async () => {
|
|
throw new Error("DB connection exploded in step.run");
|
|
});
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_throw_run", instanceId: "inst-throw-run", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-throw-run", status: "pending", name: "test_throw_run", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "failed" });
|
|
});
|
|
|
|
it("handles dynamic JSON workflow with waitForEvent step and unknown actions", async () => {
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "dynamic_wait_flow", instanceId: "inst-dyn-wait", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
// 1. Definition lookup
|
|
.mockResolvedValueOnce([
|
|
{
|
|
projectId: "p1",
|
|
name: "dynamic_wait_flow",
|
|
steps: [
|
|
{ action: "unknown_custom_action" },
|
|
{ action: "waitForEvent", event: "user.verified", options: { timeout: "2h" } },
|
|
],
|
|
},
|
|
])
|
|
// 2. Instance lookup
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-dyn-wait", status: "pending", name: "dynamic_wait_flow", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
// Waiter inserted and timer scheduled
|
|
expect(mockRedis.zadd).toHaveBeenCalled();
|
|
expect(mockDb.set).toHaveBeenCalledWith({ status: "pending" });
|
|
});
|
|
|
|
it("always releases lock in finally block", async () => {
|
|
workflowRegistry.register("test_lock_release", async (ctx) => {
|
|
await ctx.step.notify({ template: "tpl-1", user: "u-1" });
|
|
});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_lock_release", instanceId: "inst-lock-rel", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-lock-rel", status: "pending", name: "test_lock_release", input: {} },
|
|
])
|
|
.mockResolvedValueOnce([]);
|
|
|
|
await worker.process(msg);
|
|
|
|
expect(mockRedis.releaseLock).toHaveBeenCalledWith(
|
|
"lock:workflow:inst-lock-rel",
|
|
expect.any(String),
|
|
);
|
|
});
|
|
|
|
it("skips execution if workflow instance status is not pending", async () => {
|
|
workflowRegistry.register("test_skip_non_pending", async () => {});
|
|
|
|
const msg: StreamMessage = {
|
|
id: "1",
|
|
event: buildStreamEvent(
|
|
"workflow.triggered",
|
|
{ name: "test_skip_non_pending", instanceId: "inst-completed", projectId: "p1" },
|
|
"test",
|
|
"trace",
|
|
) as any,
|
|
};
|
|
|
|
mockDb.limit = vi
|
|
.fn()
|
|
.mockResolvedValueOnce([
|
|
{ id: "inst-completed", status: "completed", name: "test_skip_non_pending", input: {} },
|
|
]);
|
|
|
|
await worker.process(msg);
|
|
|
|
// Should NOT update to running
|
|
expect(mockDb.update).not.toHaveBeenCalled();
|
|
});
|
|
});
|