Add PostgreSQL-backed background worker

2026-08-17 01:02:44 -04:00
parent 9929d7321d
commit 24a1307a96
22 changed files with 629 additions and 120 deletions
@@ -1,7 +1,6 @@
 import { type NextRequest, NextResponse } from "next/server";
 import { env } from "~/env";
-import { db } from "~/server/db";
-import { generateDueRecurringInvoices } from "~/server/api/routers/recurring-invoices";
+import { scheduleDueRecurringInvoiceJobs } from "~/server/jobs/queue";

 export async function POST(req: NextRequest) {
  const authHeader = req.headers.get("authorization");
@@ -18,6 +17,6 @@ export async function POST(req: NextRequest) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

-  const generated = await generateDueRecurringInvoices(db);
-  return NextResponse.json({ generated });
+  const result = await scheduleDueRecurringInvoiceJobs();
+  return NextResponse.json(result);
 }
@@ -1,107 +1,17 @@
 import { z } from "zod";
-import { and, eq, lte } from "drizzle-orm";
+import { eq } from "drizzle-orm";
 import { createTRPCRouter, protectedProcedure } from "../trpc";
 import {
  recurringInvoices,
  recurringInvoiceItems,
-  invoices,
-  invoiceItems,
  clients,
  businesses,
 } from "~/server/db/schema";
 import { TRPCError } from "@trpc/server";
-import type { db as DbType } from "~/server/db";
-
-export function nextDueDate(schedule: string, from = new Date()): Date {
-  const d = new Date(from);
-  switch (schedule) {
-    case "weekly":    d.setDate(d.getDate() + 7); break;
-    case "biweekly":  d.setDate(d.getDate() + 14); break;
-    case "monthly":   d.setMonth(d.getMonth() + 1); break;
-    case "quarterly": d.setMonth(d.getMonth() + 3); break;
-    case "yearly":    d.setFullYear(d.getFullYear() + 1); break;
-  }
-  return d;
-}
-
-type RecurringWithItems = typeof recurringInvoices.$inferSelect & {
-  items: (typeof recurringInvoiceItems.$inferSelect)[];
-};
-
-export async function generateInvoiceFromRecurring(
-  db: typeof DbType,
-  recurring: RecurringWithItems,
-): Promise<{ id: string }> {
-  const now = new Date();
-  const invoiceNumber = `REC-${Date.now()}`;
-
-  const subtotal = recurring.items.reduce((s, i) => s + i.hours * i.rate, 0);
-  const taxAmount = (subtotal * recurring.taxRate) / 100;
-  const total = subtotal + taxAmount;
-
-  const [newInvoice] = await db
-    .insert(invoices)
-    .values({
-      invoiceNumber,
-      invoicePrefix: recurring.invoicePrefix ?? "#",
-      clientId: recurring.clientId,
-      businessId: recurring.businessId ?? null,
-      issueDate: now,
-      dueDate: nextDueDate("monthly", now),
-      status: "draft",
-      totalAmount: total,
-      taxRate: recurring.taxRate,
-      notes: recurring.notes ?? null,
-      emailMessage: recurring.emailMessage ?? null,
-      currency: recurring.currency,
-      createdById: recurring.createdById,
-    })
-    .returning({ id: invoices.id });
-
-  if (!newInvoice) throw new Error("Failed to create invoice");
-
-  if (recurring.items.length > 0) {
-    await db.insert(invoiceItems).values(
-      recurring.items.map((item, idx) => ({
-        invoiceId: newInvoice.id,
-        date: now,
-        description: item.description,
-        hours: item.hours,
-        rate: item.rate,
-        amount: item.hours * item.rate,
-        position: item.position ?? idx,
-      })),
-    );
-  }
-
-  return newInvoice;
-}
-
-export async function generateDueRecurringInvoices(db: typeof DbType): Promise<number> {
-  const now = new Date();
-  const due = await db.query.recurringInvoices.findMany({
-    where: and(
-      eq(recurringInvoices.status, "active"),
-      lte(recurringInvoices.nextDueAt, now),
-    ),
-    with: { items: true },
-  });
-
-  let generated = 0;
-  for (const rec of due) {
-    try {
-      await generateInvoiceFromRecurring(db, rec);
-      await db
-        .update(recurringInvoices)
-        .set({ lastGeneratedAt: now, nextDueAt: nextDueDate(rec.schedule, now) })
-        .where(eq(recurringInvoices.id, rec.id));
-      generated++;
-    } catch {
-      // continue on individual failures
-    }
-  }
-  return generated;
-}
+import {
+  generateInvoiceFromRecurring,
+  nextDueDate,
+} from "~/server/services/recurring-invoices";

 const scheduleEnum = z.enum(["weekly", "biweekly", "monthly", "quarterly", "yearly"]);

@@ -741,6 +741,36 @@ export const recurringInvoiceItemsRelations = relations(
  }),
 );

+// ─── Background Jobs ─────────────────────────────────────────────────────────
+
+export const backgroundJobs = createTable(
+  "background_job",
+  (d) => ({
+    id: d
+      .varchar({ length: 255 })
+      .notNull()
+      .primaryKey()
+      .$defaultFn(() => crypto.randomUUID()),
+    type: d.varchar({ length: 100 }).notNull(),
+    payload: d.jsonb().$type<Record<string, unknown>>().notNull().default({}),
+    status: d.varchar({ length: 20 }).notNull().default("pending"),
+    idempotencyKey: d.varchar({ length: 500 }).notNull().unique(),
+    runAt: d.timestamp().notNull().defaultNow(),
+    attempts: d.integer().notNull().default(0),
+    maxAttempts: d.integer().notNull().default(5),
+    lockedAt: d.timestamp(),
+    lockedBy: d.varchar({ length: 255 }),
+    lastError: d.text(),
+    completedAt: d.timestamp(),
+    createdAt: d.timestamp().notNull().defaultNow(),
+    updatedAt: d.timestamp().notNull().defaultNow(),
+  }),
+  (t) => [
+    index("background_job_status_run_at_idx").on(t.status, t.runAt),
+    index("background_job_type_status_idx").on(t.type, t.status),
+  ],
+);
+
 // ─── Time Entries ─────────────────────────────────────────────────────────────

 export const timeEntries = createTable(
@@ -0,0 +1,40 @@
+import { and, eq, lte } from "drizzle-orm";
+
+import { db } from "~/server/db";
+import { recurringInvoices } from "~/server/db/schema";
+import type { BackgroundJob } from "~/server/jobs/queue";
+import {
+  generateInvoiceFromRecurring,
+  nextDueDate,
+} from "~/server/services/recurring-invoices";
+
+export async function generateRecurringInvoice(job: BackgroundJob) {
+  const recurringInvoiceId = job.payload.recurringInvoiceId;
+  const scheduledForValue = job.payload.scheduledFor;
+  if (typeof recurringInvoiceId !== "string" || typeof scheduledForValue !== "string") {
+    throw new Error("Invalid recurring invoice job payload");
+  }
+  const scheduledFor = new Date(scheduledForValue);
+  if (Number.isNaN(scheduledFor.getTime())) throw new Error("Invalid recurring invoice job payload");
+
+  await db.transaction(async (tx) => {
+    const recurring = await tx.query.recurringInvoices.findFirst({
+      where: and(
+        eq(recurringInvoices.id, recurringInvoiceId),
+        eq(recurringInvoices.status, "active"),
+        lte(recurringInvoices.nextDueAt, scheduledFor),
+      ),
+      with: { items: true },
+    });
+    if (!recurring) return;
+
+    await generateInvoiceFromRecurring(tx, recurring);
+    await tx
+      .update(recurringInvoices)
+      .set({
+        lastGeneratedAt: new Date(),
+        nextDueAt: nextDueDate(recurring.schedule, scheduledFor),
+      })
+      .where(eq(recurringInvoices.id, recurring.id));
+  });
+}
@@ -0,0 +1,127 @@
+import { randomUUID } from "node:crypto";
+import { and, asc, eq, lte, or } from "drizzle-orm";
+
+import { db } from "~/server/db";
+import { backgroundJobs, recurringInvoices } from "~/server/db/schema";
+
+export const jobTypes = {
+  generateRecurringInvoice: "recurring_invoice.generate",
+  sendInvoice: "invoice.send_scheduled",
+  sendInvoiceReminder: "invoice.reminder.send",
+  sendPushNotification: "push_notification.send",
+  timeClockReminder: "time_clock.reminder",
+} as const;
+
+export type JobType = (typeof jobTypes)[keyof typeof jobTypes];
+export type BackgroundJob = typeof backgroundJobs.$inferSelect;
+
+export async function enqueueJob(input: {
+  type: JobType;
+  payload?: Record<string, unknown>;
+  idempotencyKey: string;
+  runAt?: Date;
+  maxAttempts?: number;
+}) {
+  const [job] = await db
+    .insert(backgroundJobs)
+    .values({
+      id: randomUUID(),
+      type: input.type,
+      payload: input.payload ?? {},
+      idempotencyKey: input.idempotencyKey,
+      runAt: input.runAt ?? new Date(),
+      maxAttempts: input.maxAttempts ?? 5,
+    })
+    .onConflictDoNothing({ target: backgroundJobs.idempotencyKey })
+    .returning();
+  return job ?? null;
+}
+
+export async function scheduleDueRecurringInvoiceJobs(now = new Date()) {
+  const due = await db.query.recurringInvoices.findMany({
+    where: and(
+      eq(recurringInvoices.status, "active"),
+      lte(recurringInvoices.nextDueAt, now),
+    ),
+  });
+  let enqueued = 0;
+  for (const recurring of due) {
+    const scheduledFor = recurring.nextDueAt.toISOString();
+    const job = await enqueueJob({
+      type: jobTypes.generateRecurringInvoice,
+      idempotencyKey: `${jobTypes.generateRecurringInvoice}:${recurring.id}:${scheduledFor}`,
+      payload: { recurringInvoiceId: recurring.id, scheduledFor },
+    });
+    if (job) enqueued++;
+  }
+  return { due: due.length, enqueued };
+}
+
+export async function claimNextJob(workerId: string) {
+  return db.transaction(async (tx) => {
+    const staleBefore = new Date(Date.now() - 5 * 60_000);
+    const [job] = await tx
+      .select()
+      .from(backgroundJobs)
+      .where(
+        and(
+          lte(backgroundJobs.runAt, new Date()),
+          or(
+            eq(backgroundJobs.status, "pending"),
+            and(
+              eq(backgroundJobs.status, "processing"),
+              lte(backgroundJobs.lockedAt, staleBefore),
+            ),
+          ),
+        ),
+      )
+      .orderBy(asc(backgroundJobs.runAt), asc(backgroundJobs.createdAt))
+      .limit(1)
+      .for("update", { skipLocked: true });
+    if (!job) return null;
+
+    const [claimed] = await tx
+      .update(backgroundJobs)
+      .set({
+        status: "processing",
+        attempts: job.attempts + 1,
+        lockedAt: new Date(),
+        lockedBy: workerId,
+        updatedAt: new Date(),
+      })
+      .where(eq(backgroundJobs.id, job.id))
+      .returning();
+    return claimed ?? null;
+  });
+}
+
+export async function completeJob(id: string) {
+  await db
+    .update(backgroundJobs)
+    .set({
+      status: "completed",
+      completedAt: new Date(),
+      lockedAt: null,
+      lockedBy: null,
+      lastError: null,
+      updatedAt: new Date(),
+    })
+    .where(eq(backgroundJobs.id, id));
+}
+
+export async function failJob(job: BackgroundJob, error: unknown) {
+  const terminal = job.attempts >= job.maxAttempts;
+  const retryDelayMs = Math.min(60 * 60_000, 2 ** Math.max(0, job.attempts - 1) * 15_000);
+  await db
+    .update(backgroundJobs)
+    .set({
+      status: terminal ? "failed" : "pending",
+      runAt: terminal ? job.runAt : new Date(Date.now() + retryDelayMs),
+      lockedAt: null,
+      lockedBy: null,
+      lastError: error instanceof Error ? error.message : "Unknown error",
+      updatedAt: new Date(),
+    })
+    .where(eq(backgroundJobs.id, job.id));
+  return terminal;
+}
@@ -0,0 +1,79 @@
+import type { db as DbType } from "~/server/db";
+import { invoiceItems, invoices } from "~/server/db/schema";
+import type {
+  recurringInvoiceItems,
+  recurringInvoices,
+} from "~/server/db/schema";
+
+export function nextDueDate(schedule: string, from = new Date()): Date {
+  const date = new Date(from);
+  switch (schedule) {
+    case "weekly":
+      date.setDate(date.getDate() + 7);
+      break;
+    case "biweekly":
+      date.setDate(date.getDate() + 14);
+      break;
+    case "monthly":
+      date.setMonth(date.getMonth() + 1);
+      break;
+    case "quarterly":
+      date.setMonth(date.getMonth() + 3);
+      break;
+    case "yearly":
+      date.setFullYear(date.getFullYear() + 1);
+      break;
+  }
+  return date;
+}
+
+type RecurringWithItems = typeof recurringInvoices.$inferSelect & {
+  items: (typeof recurringInvoiceItems.$inferSelect)[];
+};
+
+export async function generateInvoiceFromRecurring(
+  db: Pick<typeof DbType, "insert">,
+  recurring: RecurringWithItems,
+): Promise<{ id: string }> {
+  const now = new Date();
+  const invoiceNumber = `REC-${Date.now()}`;
+  const subtotal = recurring.items.reduce((sum, item) => sum + item.hours * item.rate, 0);
+  const taxAmount = (subtotal * recurring.taxRate) / 100;
+
+  const [newInvoice] = await db
+    .insert(invoices)
+    .values({
+      invoiceNumber,
+      invoicePrefix: recurring.invoicePrefix ?? "#",
+      clientId: recurring.clientId,
+      businessId: recurring.businessId ?? null,
+      issueDate: now,
+      dueDate: nextDueDate("monthly", now),
+      status: "draft",
+      totalAmount: subtotal + taxAmount,
+      taxRate: recurring.taxRate,
+      notes: recurring.notes ?? null,
+      emailMessage: recurring.emailMessage ?? null,
+      currency: recurring.currency,
+      createdById: recurring.createdById,
+    })
+    .returning({ id: invoices.id });
+
+  if (!newInvoice) throw new Error("Failed to create invoice");
+
+  if (recurring.items.length > 0) {
+    await db.insert(invoiceItems).values(
+      recurring.items.map((item, index) => ({
+        invoiceId: newInvoice.id,
+        date: now,
+        description: item.description,
+        hours: item.hours,
+        rate: item.rate,
+        amount: item.hours * item.rate,
+        position: item.position ?? index,
+      })),
+    );
+  }
+
+  return newInvoice;
+}