test: add unit tests for delivery circuit breaker, fallback chains, and console transport

- Add test coverage for DeliveryWorker circuit breaker isolation, consecutive failure trips, and probe recovery

- Add test coverage for fallback chain exhaustion and terminal NonRetryableError assertions

- Add comprehensive test suite for ConsoleTransport dev provider

tests/providers.test.ts

@@ -1,6 +1,7 @@
-import { describe, it, expect, vi, beforeEach } from "vitest";
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 import { ResendTransport } from "../packages/provider-resend/src/index.js";
 import { FcmTransport } from "../packages/provider-fcm/src/index.js";
+import { ConsoleTransport } from "../packages/provider-console/src/index.js";
 
 // Mock Resend SDK
 const mockSendEmail = vi.fn();
@@ -406,3 +407,88 @@ describe("TransportRegistry", () => {
     expect(transportRegistry.get("email")).toBe(mockTransport2);
   });
 });
+
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe("ConsoleTransport (Development Provider)", () => {
+  /** The transport's only side effects are stdout and the optional logger. */
+  let logSpy: ReturnType<typeof vi.spyOn>;
+
+  beforeEach(() => {
+    logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
+  });
+
+  afterEach(() => {
+    logSpy.mockRestore();
+  });
+
+  const task = (over: Record<string, unknown> = {}): any => ({
+    taskId: "task-uuid-1",
+    enrichedEventId: "enriched-evt-1",
+    recipientId: "usr-1",
+    destination: "device-token-1",
+    priority: "normal",
+    renderedContent: { content: { body: "Hello World" } },
+    ...over,
+  });
+
+  it("reports a success carrying a console-prefixed provider message id", async () => {
+    const result = await new ConsoleTransport().send(task());
+
+    expect(result.success).toBe(true);
+    expect(result.providerMessageId).toMatch(/^console-\d+$/);
+  });
+
+  it("defaults to the push channel when none is named", () => {
+    expect(new ConsoleTransport().channel).toBe("push");
+  });
+
+  it("takes the channel it is given, so it can stand in for email or sms", () => {
+    expect(new ConsoleTransport({ channel: "email" }).channel).toBe("email");
+  });
+
+  it("carries limits through, and leaves them unset when none are given", () => {
+    const limits = { limit: 10, windowSeconds: 60 };
+
+    expect(new ConsoleTransport({ limits }).limits).toEqual(limits);
+    expect(new ConsoleTransport().limits).toBeUndefined();
+  });
+
+  it("prints the parts of the task a developer is watching for", async () => {
+    await new ConsoleTransport().send(task({ renderedContent: { content: { body: "Ship it" } } }));
+
+    expect(logSpy).toHaveBeenCalledTimes(1);
+    const printed = logSpy.mock.calls[0]![0] as string;
+
+    expect(printed).toContain("device-token-1"); // destination
+    expect(printed).toContain("usr-1"); // recipient
+    expect(printed).toContain("normal"); // priority
+    expect(printed).toContain("task-uuid-1"); // taskId
+    expect(printed).toContain(JSON.stringify({ body: "Ship it" })); // content
+  });
+
+  it("logs the delivery with the ids a caller would correlate on", async () => {
+    const logger: any = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() };
+
+    const result = await new ConsoleTransport({ channel: "sms", logger }).send(task());
+
+    expect(logger.info).toHaveBeenCalledWith(
+      {
+        taskId: "task-uuid-1",
+        channel: "sms",
+        destination: "device-token-1",
+        providerMessageId: result.providerMessageId,
+      },
+      "push delivered (console transport)",
+    );
+  });
+
+  it("delivers without a logger, since one is optional", async () => {
+    const transport = new ConsoleTransport();
+
+    await expect(transport.send(task())).resolves.toEqual({
+      success: true,
+      providerMessageId: expect.stringMatching(/^console-\d+$/),
+    });
+  });
+});

tests/workers.test.ts

@@ -2,7 +2,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 import { AiWorker, PermanentAiError } from "@/services/ai/main.js";
 import { DeliveryWorker } from "@/services/delivery/main.js";
 import { PreferenceRepository } from "@/repositories/index.js";
-import { BaseWorker } from "@/workers/index.js";
+import { BaseWorker, NonRetryableError } from "@/workers/index.js";
 import type { StreamMessage } from "@/queue/index.js";
 import { createMockDb, type MockDb } from "./helpers/mock-db.js";
 import { suppressions } from "@/db/schema.js";
@@ -1241,6 +1241,225 @@ describe("DeliveryWorker", () => {
       expect(second).not.toHaveBeenCalled();
     });
   });
+
+  // ── Circuit breaker ───────────────────────────────────────────────────────
+
+  /**
+   * The worker keys one breaker per `${channel}:${transport.constructor.name}`,
+   * so these transports are real classes. Two object literals would both report
+   * the name "Object" and quietly share a single breaker.
+   */
+  describe("circuit breaker", () => {
+    class FlakyTransport {
+      send = vi.fn().mockRejectedValue(new Error("Provider down"));
+    }
+    class HealthyTransport {
+      send = vi.fn().mockResolvedValue({ success: true, providerMessageId: "ok-1" });
+    }
+
+    /** One delivery, absorbing the NonRetryableError a chainless task ends in. */
+    const attempt = (msg: unknown) => worker.process(msg as any).catch((e) => e);
+
+    let nowSpy: ReturnType<typeof vi.spyOn> | undefined;
+    afterEach(() => {
+      nowSpy?.mockRestore();
+      nowSpy = undefined;
+    });
+
+    it("stops calling a transport once five consecutive failures trip the breaker", async () => {
+      const flaky = new FlakyTransport();
+      mockTransportRegistry.getAll.mockReturnValue([flaky]);
+
+      for (let i = 0; i < 5; i++) await attempt(dispatched());
+      expect(flaky.send).toHaveBeenCalledTimes(5);
+
+      await attempt(dispatched());
+
+      // The sixth task never reached the provider — that is the whole point.
+      expect(flaky.send).toHaveBeenCalledTimes(5);
+    });
+
+    it("reports an open breaker as a delivery failure instead of letting it escape", async () => {
+      mockTransportRegistry.getAll.mockReturnValue([new FlakyTransport()]);
+
+      for (let i = 0; i < 5; i++) await attempt(dispatched());
+      const err = await attempt(dispatched());
+
+      // A short-circuit is still a delivery failure: it must reach the emitter
+      // and come back as NonRetryableError, or the task retries forever against
+      // a provider the worker has already given up on.
+      expect(err).toBeInstanceOf(Error);
+      expect((err as Error).message).toBe("Circuit breaker is OPEN");
+      expect(mockGlobalEmitter.emit).toHaveBeenLastCalledWith(
+        "delivery:failed",
+        "123e4567-e89b-12d3-a456-426614174001",
+        "Circuit breaker is OPEN",
+        "email",
+        "123e4567-e89b-12d3-a456-426614174000",
+      );
+    });
+
+    it("keeps one provider's breaker off another's, so a healthy provider still delivers", async () => {
+      const flaky = new FlakyTransport();
+      const healthy = new HealthyTransport();
+      mockTransportRegistry.getAll.mockReturnValue([flaky, healthy]);
+
+      for (let i = 0; i < 6; i++) await attempt(dispatched());
+
+      expect(flaky.send).toHaveBeenCalledTimes(5); // tripped, then skipped
+      expect(healthy.send).toHaveBeenCalledTimes(6); // never affected
+      expect(mockGlobalEmitter.emit).toHaveBeenLastCalledWith(
+        "delivery:delivered",
+        "123e4567-e89b-12d3-a456-426614174001",
+        "ok-1",
+        "email",
+        "123e4567-e89b-12d3-a456-426614174000",
+      );
+    });
+
+    it("keeps a channel's breaker off the same transport on another channel", async () => {
+      const flaky = new FlakyTransport();
+      mockTransportRegistry.getAll.mockReturnValue([flaky]);
+
+      for (let i = 0; i < 6; i++) await attempt(dispatched());
+      expect(flaky.send).toHaveBeenCalledTimes(5); // email is now open
+
+      await attempt(dispatched({ channel: "sms", destination: "+15551234567" }));
+
+      // sms keys a different breaker, so it gets its own five attempts.
+      expect(flaky.send).toHaveBeenCalledTimes(6);
+    });
+
+    it("resets the count on a success, so intermittent failures never trip it", async () => {
+      const intermittent = new FlakyTransport();
+      intermittent.send
+        .mockRejectedValue(new Error("Provider down"))
+        .mockResolvedValueOnce({ success: true, providerMessageId: "ok-2" });
+      mockTransportRegistry.getAll.mockReturnValue([intermittent]);
+
+      // One success on the very first call, then eight straight failures. The
+      // threshold counts consecutive failures, so four-then-four never trips.
+      await attempt(dispatched());
+      for (let i = 0; i < 4; i++) await attempt(dispatched());
+      intermittent.send.mockResolvedValueOnce({ success: true, providerMessageId: "ok-3" });
+      await attempt(dispatched());
+      for (let i = 0; i < 4; i++) await attempt(dispatched());
+
+      expect(intermittent.send).toHaveBeenCalledTimes(10);
+    });
+
+    it("lets a single probe through once the reset window has passed", async () => {
+      const flaky = new FlakyTransport();
+      mockTransportRegistry.getAll.mockReturnValue([flaky]);
+
+      for (let i = 0; i < 6; i++) await attempt(dispatched());
+      expect(flaky.send).toHaveBeenCalledTimes(5);
+
+      // The provider recovers, but the breaker holds it out for a full 30s.
+      flaky.send.mockResolvedValue({ success: true, providerMessageId: "recovered" });
+      await attempt(dispatched());
+      expect(flaky.send).toHaveBeenCalledTimes(5);
+
+      nowSpy = vi.spyOn(Date, "now").mockReturnValue(Date.now() + 31_000);
+      await attempt(dispatched());
+
+      expect(flaky.send).toHaveBeenCalledTimes(6);
+      expect(mockGlobalEmitter.emit).toHaveBeenLastCalledWith(
+        "delivery:delivered",
+        "123e4567-e89b-12d3-a456-426614174001",
+        "recovered",
+        "email",
+        "123e4567-e89b-12d3-a456-426614174000",
+      );
+    });
+  });
+
+  // ── Fallback chain exhaustion ─────────────────────────────────────────────
+
+  /**
+   * Rolling over to the next channel is well covered; running *out* of channels
+   * is the other half. Every path here has to end in NonRetryableError, because
+   * a retryable throw would send an already-exhausted task around again.
+   */
+  describe("fallback chain exhaustion", () => {
+    const allProvidersFail = () =>
+      mockTransportRegistry.getAll.mockReturnValue([
+        { send: vi.fn().mockRejectedValue(new Error("Provider down")) },
+      ]);
+
+    it("ends the task when the only channel fails with nothing to fall back to", async () => {
+      allProvidersFail();
+
+      await expect(worker.process(dispatched() as any)).rejects.toBeInstanceOf(NonRetryableError);
+
+      expect(mockEnrichedProducers.normal.publish).not.toHaveBeenCalled();
+      expect(mockGlobalEmitter.emit).toHaveBeenCalledWith(
+        "delivery:failed",
+        "123e4567-e89b-12d3-a456-426614174001",
+        "Provider down",
+        "email",
+        "123e4567-e89b-12d3-a456-426614174000",
+      );
+    });
+
+    it("treats an empty fallbackChain as no chain at all", async () => {
+      allProvidersFail();
+
+      await expect(
+        worker.process(dispatched({ fallbackChain: [], recipient }) as any),
+      ).rejects.toBeInstanceOf(NonRetryableError);
+
+      expect(mockEnrichedProducers.normal.publish).not.toHaveBeenCalled();
+    });
+
+    it("does not roll over when a chain is present but the recipient is missing", async () => {
+      allProvidersFail();
+
+      // The rollover rebuilds an enriched payload, which needs the recipient.
+      // Without one the chain is inert, so the task has to fail terminally
+      // rather than look like it was handed on.
+      await expect(
+        worker.process(dispatched({ fallbackChain: ["sms", "push"] }) as any),
+      ).rejects.toBeInstanceOf(NonRetryableError);
+
+      expect(mockEnrichedProducers.normal.publish).not.toHaveBeenCalled();
+    });
+
+    it("clears the chain on the final hop instead of passing an empty array on", async () => {
+      allProvidersFail();
+
+      await worker.process(dispatched({ fallbackChain: ["sms"], recipient }) as any);
+
+      const published = mockEnrichedProducers.normal.publish.mock.calls[0]![0];
+      expect(published.payload.channel).toBe("sms");
+      // Not [] — an empty array would read as "a chain exists" to the next hop.
+      expect(published.payload.fallbackChain).toBeUndefined();
+    });
+
+    it("stops at the end of the chain rather than rolling over forever", async () => {
+      allProvidersFail();
+
+      // Hop one: email fails and hands the task to sms with the chain spent.
+      await worker.process(dispatched({ fallbackChain: ["sms"], recipient }) as any);
+      const next = mockEnrichedProducers.normal.publish.mock.calls[0]![0].payload;
+      expect(next.channel).toBe("sms");
+
+      // Hop two: that sms task fails too, and now there is nowhere left to go.
+      mockEnrichedProducers.normal.publish.mockClear();
+      await expect(
+        worker.process(
+          dispatched({
+            channel: next.channel,
+            fallbackChain: next.fallbackChain,
+            destination: "+15551234567",
+            recipient,
+          }) as any,
+        ),
+      ).rejects.toBeInstanceOf(NonRetryableError);
+
+      expect(mockEnrichedProducers.normal.publish).not.toHaveBeenCalled();
+    });
+  });
 });
 
 import { NotifkitServer } from "@/server.js";