Add scheduled invoice delivery

apps/worker/README.md

@@ -2,10 +2,12 @@
 
 The worker is a separate Bun process backed by the same PostgreSQL database as the web app. It follows the Racetix worker model: a durable database outbox, a lightweight scheduler, and horizontally safe polling with `FOR UPDATE SKIP LOCKED`.
 
-Currently it schedules and generates due recurring invoices. New asynchronous workflows should enqueue a typed job through `apps/web/src/server/jobs/queue.ts` and add a handler in `src/index.ts`.
+It generates due recurring invoices and delivers scheduled invoice emails. New asynchronous workflows should enqueue a typed job through `apps/web/src/server/jobs/queue.ts` and add a handler in `src/index.ts`.
 
 Jobs have an idempotency key, scheduled run time, bounded exponential retries, and stale-lock recovery. Multiple worker replicas can run safely. Timer elapsed time is still derived from `startedAt`; the worker should only send time-clock reminders, never increment a counter every second.
 
+Scheduled sends store an absolute UTC instant plus the IANA timezone selected by the client. The worker calls the app's secret-protected internal delivery endpoint through `APP_INTERNAL_URL`; that endpoint passes the job idempotency key to Resend, so a retry cannot send the same invoice twice.
+
 ```bash
 # Uses the web app's .env/.env.local files
 bun run dev
@@ -15,3 +17,4 @@ bun run start
 ```
 
 `WORKER_POLL_MS` defaults to 2000 and `WORKER_SCHEDULE_MS` defaults to 60000.
+`APP_INTERNAL_URL` defaults to `http://app:3000` in Compose, and `CRON_SECRET` authenticates worker requests to the app.

apps/worker/src/index.ts

@@ -6,9 +6,11 @@ import {
   completeJob,
   failJob,
   jobTypes,
+  markScheduledInvoiceJobFailed,
   scheduleDueRecurringInvoiceJobs,
   type BackgroundJob,
 } from "../../web/src/server/jobs/queue";
+import { sendScheduledInvoice } from "./send-invoice";
 
 const workerId = `beenvoice-worker:${randomUUID()}`;
 const pollMs = Number(process.env.WORKER_POLL_MS ?? 2_000);
@@ -17,7 +19,11 @@ let stopping = false;
 let working = false;
 let scheduling = false;
 
-function log(level: "info" | "error", event: string, fields: Record<string, unknown> = {}) {
+function log(
+  level: "info" | "error",
+  event: string,
+  fields: Record<string, unknown> = {},
+) {
   const record = JSON.stringify({
     timestamp: new Date().toISOString(),
     level,
@@ -59,6 +65,10 @@ async function handleJob(job: BackgroundJob) {
     await generateRecurringInvoice(job);
     return;
   }
+  if (job.type === jobTypes.sendInvoice) {
+    await sendScheduledInvoice(job);
+    return;
+  }
   throw new Error(`No handler registered for ${job.type}`);
 }
 
@@ -72,9 +82,16 @@ async function drainJobs() {
       try {
         await handleJob(job);
         await completeJob(job.id);
-        log("info", "job.completed", { jobId: job.id, jobType: job.type, attempts: job.attempts });
+        log("info", "job.completed", {
+          jobId: job.id,
+          jobType: job.type,
+          attempts: job.attempts,
+        });
       } catch (error) {
         const terminal = await failJob(job, error);
+        if (terminal && job.type === jobTypes.sendInvoice) {
+          await markScheduledInvoiceJobFailed(job);
+        }
         log("error", "job.failed", {
           jobId: job.id,
           jobType: job.type,

apps/worker/src/send-invoice.ts

@@ -0,0 +1,41 @@
+import type { BackgroundJob } from "../../web/src/server/jobs/queue";
+
+function requiredPayloadString(job: BackgroundJob, key: string): string {
+  const value = job.payload[key];
+  if (typeof value !== "string" || !value) {
+    throw new Error(`Invalid scheduled invoice job payload: ${key}`);
+  }
+  return value;
+}
+
+export async function sendScheduledInvoice(job: BackgroundJob) {
+  const appUrl =
+    process.env.APP_INTERNAL_URL?.replace(/\/$/, "") ?? "http://app:3000";
+  const secret = process.env.CRON_SECRET;
+  if (!secret)
+    throw new Error("CRON_SECRET is required for scheduled invoice delivery");
+
+  const response = await fetch(`${appUrl}/api/internal/jobs/send-invoice`, {
+    method: "POST",
+    headers: {
+      Authorization: `Bearer ${secret}`,
+      "Content-Type": "application/json",
+    },
+    body: JSON.stringify({
+      ...job.payload,
+      jobId: job.id,
+      idempotencyKey: job.idempotencyKey,
+      invoiceId: requiredPayloadString(job, "invoiceId"),
+      actorUserId: requiredPayloadString(job, "actorUserId"),
+    }),
+  });
+  const result = (await response.json().catch(() => ({}))) as {
+    error?: string;
+  };
+  if (!response.ok) {
+    throw new Error(
+      result.error ?? `Invoice delivery request failed (${response.status})`,
+    );
+  }
+  return result;
+}

apps/worker/tests/scheduled-invoice-send.test.ts

@@ -0,0 +1,88 @@
+/// <reference types="bun" />
+
+import { afterEach, describe, expect, mock, test } from "bun:test";
+
+import type { BackgroundJob } from "../../web/src/server/jobs/queue";
+import { sendScheduledInvoice } from "../src/send-invoice";
+
+const originalFetch = globalThis.fetch;
+const originalAppUrl = process.env.APP_INTERNAL_URL;
+const originalSecret = process.env.CRON_SECRET;
+
+function scheduledJob(): BackgroundJob {
+  const now = new Date("2026-08-17T16:00:00.000Z");
+  return {
+    id: "job-1",
+    type: "invoice.send_scheduled",
+    payload: {
+      invoiceId: "invoice-1",
+      actorUserId: "user-1",
+      customMessage: "Thanks!",
+      timeZone: "America/New_York",
+    },
+    status: "processing",
+    idempotencyKey: "invoice.send_scheduled:invoice-1:once",
+    runAt: now,
+    attempts: 1,
+    maxAttempts: 5,
+    lockedAt: now,
+    lockedBy: "worker-1",
+    lastError: null,
+    completedAt: null,
+    createdAt: now,
+    updatedAt: now,
+  };
+}
+
+afterEach(() => {
+  globalThis.fetch = originalFetch;
+  if (originalAppUrl === undefined) delete process.env.APP_INTERNAL_URL;
+  else process.env.APP_INTERNAL_URL = originalAppUrl;
+  if (originalSecret === undefined) delete process.env.CRON_SECRET;
+  else process.env.CRON_SECRET = originalSecret;
+});
+
+describe("scheduled invoice delivery", () => {
+  test("calls the internal app endpoint with auth and an idempotency key", async () => {
+    process.env.APP_INTERNAL_URL = "http://app:3000/";
+    process.env.CRON_SECRET = "worker-secret";
+    const fetchMock = mock(
+      async (_url: string | URL | Request, _request?: RequestInit) =>
+        Response.json({ success: true, emailId: "email-1" }),
+    );
+    globalThis.fetch = fetchMock as unknown as typeof fetch;
+
+    await sendScheduledInvoice(scheduledJob());
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const [url, request] = fetchMock.mock.calls[0]!;
+    expect(url).toBe("http://app:3000/api/internal/jobs/send-invoice");
+    expect(request?.headers).toEqual({
+      Authorization: "Bearer worker-secret",
+      "Content-Type": "application/json",
+    });
+    expect(JSON.parse(String(request?.body))).toMatchObject({
+      jobId: "job-1",
+      invoiceId: "invoice-1",
+      actorUserId: "user-1",
+      idempotencyKey: "invoice.send_scheduled:invoice-1:once",
+      timeZone: "America/New_York",
+    });
+  });
+
+  test("surfaces retryable delivery errors to the worker", async () => {
+    process.env.APP_INTERNAL_URL = "http://app:3000";
+    process.env.CRON_SECRET = "worker-secret";
+    globalThis.fetch = mock(
+      async (_url: string | URL | Request, _request?: RequestInit) =>
+        Response.json(
+          { error: "Email service is unavailable" },
+          { status: 503 },
+        ),
+    ) as unknown as typeof fetch;
+
+    await expect(sendScheduledInvoice(scheduledJob())).rejects.toThrow(
+      "Email service is unavailable",
+    );
+  });
+});