diff --git a/Dockerfile b/Dockerfile index f8c2eab..dd12fdf 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,10 +6,19 @@ WORKDIR /app FROM base AS install COPY package.json bun.lock ./ COPY apps/web/package.json apps/web/package.json +COPY apps/worker/package.json apps/worker/package.json COPY apps/mobile/package.json apps/mobile/package.json COPY packages/domain/package.json packages/domain/package.json RUN bun install --frozen-lockfile --filter @beenvoice/web +FROM base AS worker-install +COPY package.json bun.lock ./ +COPY apps/web/package.json apps/web/package.json +COPY apps/worker/package.json apps/worker/package.json +COPY apps/mobile/package.json apps/mobile/package.json +COPY packages/domain/package.json packages/domain/package.json +RUN bun install --frozen-lockfile --filter @beenvoice/worker + # Next's production build runs under Node because Bun can fail during the # page-data worker phase on Linux arm64. Dependencies still come from Bun. FROM node:22-bookworm-slim AS build @@ -55,3 +64,21 @@ USER bun EXPOSE 3000 WORKDIR /app/apps/web CMD ["sh", "-c", "bun src/server/db/migrate.ts && bun run start"] + +FROM base AS worker +ENV NODE_ENV=production \ + NEXT_TELEMETRY_DISABLED=1 + +COPY --from=worker-install /app ./ +COPY apps/web/src ./apps/web/src +COPY apps/web/tsconfig.json ./apps/web/tsconfig.json +COPY apps/worker ./apps/worker +COPY packages/domain ./packages/domain +RUN ln -s ../worker/node_modules apps/web/node_modules + +USER bun +WORKDIR /app/apps/worker +CMD ["bun", "run", "start"] + +# Keep the web application as the default Dockerfile target. +FROM release AS final diff --git a/README.md b/README.md index c75d3d7..749de11 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,8 @@ Beenvoice is a freelancer and small-business invoicing platform with a Next.js w beenvoice/ ├── apps/ │ ├── web/ # Next.js dashboard, tRPC API, PostgreSQL/Drizzle -│ └── mobile/ # Expo Router mobile app and iOS widgets +│ ├── mobile/ # Expo Router mobile app and iOS widgets +│ └── worker/ # PostgreSQL-backed scheduler and background jobs ├── packages/ │ └── domain/ # Platform-neutral shared rules and parsing ├── Dockerfile @@ -32,7 +33,7 @@ bun run --filter @beenvoice/web db:push bun run dev ``` -`bun run dev` starts Next.js on port 3000 and Expo Metro on port 8082. For a physical iPhone, set `EXPO_PUBLIC_API_URL` in `apps/mobile/.env` to a host the device can reach and configure the web app's canonical/auth URLs consistently. +`bun run dev` starts Next.js on port 3000, Expo Metro on port 8082, and the background worker. For a physical iPhone, set `EXPO_PUBLIC_API_URL` in `apps/mobile/.env` to a host the device can reach and configure the web app's canonical/auth URLs consistently. Useful workspace commands: @@ -61,7 +62,7 @@ git pull ./scripts/docker-deploy.sh ``` -The root Dockerfile installs the frozen Bun workspace lockfile, builds the Next.js app under Node, and runs migrations plus the web server under Bun. +The root Dockerfile installs the frozen Bun workspace lockfile and exposes separate `final` (web) and `worker` targets. The Compose stack starts both; the web container runs migrations before serving requests and the worker tolerates that short startup race. ## Documentation @@ -69,6 +70,7 @@ The root Dockerfile installs the frozen Bun workspace lockfile, builds the Next. - [Web architecture](./apps/web/docs/ARCHITECTURE.md) - [Mobile setup](./apps/mobile/README.md) - [Mobile architecture](./apps/mobile/docs/ARCHITECTURE.md) +- [Worker architecture](./apps/worker/README.md) - [Shared domain package](./packages/domain/README.md) ## Product concepts diff --git a/apps/web/README.md b/apps/web/README.md index 62e6ce2..3abae30 100644 --- a/apps/web/README.md +++ b/apps/web/README.md @@ -145,15 +145,14 @@ App listens on `${WEB_PORT:-${PORT:-3000}}` on the host (container port is alway ### Scheduled recurring invoices -The app container does not run a cron daemon. It starts the web server with -`bun migrate.ts && bun run start`, and recurring invoice generation only happens -when something calls `POST /api/cron/generate-recurring` with -`Authorization: Bearer $CRON_SECRET`. +The Compose stack includes a dedicated PostgreSQL-backed worker. It discovers due +recurring invoices every minute, enqueues idempotent jobs, and processes them with +bounded retries and stale-lock recovery. No Coolify scheduled task or Redis +service is required. -- **Coolify deploys:** use a Coolify scheduled task to call the endpoint. -- **Full Docker deploys:** use host cron, a small scheduler sidecar, or an - external scheduler to call - `http://localhost:${WEB_PORT:-${PORT:-3000}}/api/cron/generate-recurring`. +`POST /api/cron/generate-recurring` remains available as an optional authenticated +"schedule now" hook. It only enqueues due work; invoice generation stays in the +worker. ### 3. Updating an existing deploy @@ -170,7 +169,7 @@ git pull | `../../scripts/docker-deploy.sh` or root `docker compose up -d --build` | Yes | Yes — on app container start | | `docker compose restart app` | No | Yes — migrate runs again (no-op if up to date) | -Prune old app images occasionally: `docker image prune -f` (or remove specific `beenvoice:*` tags). +Prune old app and worker images occasionally: `docker image prune -f` (or remove specific `beenvoice:*` / `beenvoice-worker:*` tags). To verify migration files match the journal before deploy: `bun run db:verify-journal`. diff --git a/apps/web/docs/ARCHITECTURE.md b/apps/web/docs/ARCHITECTURE.md index 04b44ac..70f47c2 100644 --- a/apps/web/docs/ARCHITECTURE.md +++ b/apps/web/docs/ARCHITECTURE.md @@ -77,7 +77,7 @@ Root: `src/server/api/root.ts`. All routers use Zod input validation. | `payments` | `routers/payments.ts` | getByInvoice, create, delete | | `expenses` | `routers/expenses.ts` | getAll, getById, create, update, delete | | `invoiceTemplates` | `routers/invoiceTemplates.ts` | CRUD by template type | -| `recurringInvoices` | `routers/recurring-invoices.ts` | CRUD, pause/resume, generateNow; cron helper `generateDueRecurringInvoices` | +| `recurringInvoices` | `routers/recurring-invoices.ts` | CRUD, pause/resume, generateNow; due generation runs through the worker | | `timeEntries` | `routers/time-entries.ts` | getAll, getRunning, clockIn, updateRunning, clockOut, create, update, delete, getSummary | | `dashboard` | `routers/dashboard.ts` | getStats | | `email` | `routers/email.ts` | sendInvoice | @@ -183,10 +183,10 @@ Validated in `src/env.js`. See `.env.example`. | File | Use | |------|-----| -| Root `docker-compose.yml` | Deploy: `app` + `db` + Garage; use `apps/web/.env` | +| Root `docker-compose.yml` | Deploy: `app` + `worker` + `db` + Garage; use `apps/web/.env` | | Root `docker-compose.dev.yml` | Local dev: Postgres + Garage | -The app image is built from the root `Dockerfile`. Container startup runs the web migration script and then `next start` on port 3000. Docker builds run `next build` on Node 22 (not Bun) to avoid arm64 worker crashes; runtime stays on Bun. +The root `Dockerfile` has separate web and worker targets. The web container runs migrations and then `next start` on port 3000. The Bun worker uses PostgreSQL as its durable queue, polls with row locking, and schedules recurring invoice generation without Redis or an external cron. Docker builds run `next build` on Node 22 (not Bun) to avoid arm64 worker crashes; runtime stays on Bun. Set `BETTER_AUTH_URL` and `NEXT_PUBLIC_APP_URL` to the public hostname before deploy. Rebuild the image when changing `NEXT_PUBLIC_*` build-time vars. diff --git a/apps/web/drizzle/0029_background_jobs.sql b/apps/web/drizzle/0029_background_jobs.sql new file mode 100644 index 0000000..571cd53 --- /dev/null +++ b/apps/web/drizzle/0029_background_jobs.sql @@ -0,0 +1,21 @@ +CREATE TABLE IF NOT EXISTS "beenvoice_background_job" ( + "id" varchar(255) PRIMARY KEY NOT NULL, + "type" varchar(100) NOT NULL, + "payload" jsonb DEFAULT '{}'::jsonb NOT NULL, + "status" varchar(20) DEFAULT 'pending' NOT NULL, + "idempotencyKey" varchar(500) NOT NULL, + "runAt" timestamp DEFAULT now() NOT NULL, + "attempts" integer DEFAULT 0 NOT NULL, + "maxAttempts" integer DEFAULT 5 NOT NULL, + "lockedAt" timestamp, + "lockedBy" varchar(255), + "lastError" text, + "completedAt" timestamp, + "createdAt" timestamp DEFAULT now() NOT NULL, + "updatedAt" timestamp DEFAULT now() NOT NULL, + CONSTRAINT "beenvoice_background_job_idempotencyKey_unique" UNIQUE("idempotencyKey") +); +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "background_job_status_run_at_idx" ON "beenvoice_background_job" USING btree ("status", "runAt"); +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "background_job_type_status_idx" ON "beenvoice_background_job" USING btree ("type", "status"); diff --git a/apps/web/drizzle/meta/_journal.json b/apps/web/drizzle/meta/_journal.json index 3539d1f..1fd9805 100644 --- a/apps/web/drizzle/meta/_journal.json +++ b/apps/web/drizzle/meta/_journal.json @@ -204,6 +204,13 @@ "when": 1786766968000, "tag": "0028_enable_public_demo_password", "breakpoints": true + }, + { + "idx": 29, + "version": "7", + "when": 1786941568000, + "tag": "0029_background_jobs", + "breakpoints": true } ] } diff --git a/apps/web/src/app/api/cron/generate-recurring/route.ts b/apps/web/src/app/api/cron/generate-recurring/route.ts index ea95527..fff90fc 100644 --- a/apps/web/src/app/api/cron/generate-recurring/route.ts +++ b/apps/web/src/app/api/cron/generate-recurring/route.ts @@ -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); } diff --git a/apps/web/src/server/api/routers/recurring-invoices.ts b/apps/web/src/server/api/routers/recurring-invoices.ts index c1229e8..54f9e23 100644 --- a/apps/web/src/server/api/routers/recurring-invoices.ts +++ b/apps/web/src/server/api/routers/recurring-invoices.ts @@ -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 { - 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"]); diff --git a/apps/web/src/server/db/schema.ts b/apps/web/src/server/db/schema.ts index 4742f7c..3eefcb7 100644 --- a/apps/web/src/server/db/schema.ts +++ b/apps/web/src/server/db/schema.ts @@ -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>().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( diff --git a/apps/web/src/server/jobs/handlers/recurring-invoice.ts b/apps/web/src/server/jobs/handlers/recurring-invoice.ts new file mode 100644 index 0000000..4fc532e --- /dev/null +++ b/apps/web/src/server/jobs/handlers/recurring-invoice.ts @@ -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)); + }); +} diff --git a/apps/web/src/server/jobs/queue.ts b/apps/web/src/server/jobs/queue.ts new file mode 100644 index 0000000..aa760a4 --- /dev/null +++ b/apps/web/src/server/jobs/queue.ts @@ -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; + 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; +} diff --git a/apps/web/src/server/services/recurring-invoices.ts b/apps/web/src/server/services/recurring-invoices.ts new file mode 100644 index 0000000..557a0eb --- /dev/null +++ b/apps/web/src/server/services/recurring-invoices.ts @@ -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, + 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; +} diff --git a/apps/worker/README.md b/apps/worker/README.md new file mode 100644 index 0000000..41e9518 --- /dev/null +++ b/apps/worker/README.md @@ -0,0 +1,17 @@ +# Beenvoice worker + +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`. + +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. + +```bash +# Uses the web app's .env/.env.local files +bun run dev + +# Production environment is injected by Docker Compose/Coolify +bun run start +``` + +`WORKER_POLL_MS` defaults to 2000 and `WORKER_SCHEDULE_MS` defaults to 60000. diff --git a/apps/worker/package.json b/apps/worker/package.json new file mode 100644 index 0000000..2a1fb10 --- /dev/null +++ b/apps/worker/package.json @@ -0,0 +1,26 @@ +{ + "name": "@beenvoice/worker", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "dev": "bun --env-file=../web/.env --env-file=../web/.env.local --watch src/index.ts", + "start": "bun src/index.ts", + "build": "tsc --noEmit", + "typecheck": "tsc --noEmit", + "lint": "tsc --noEmit", + "test": "bun test" + }, + "dependencies": { + "@t3-oss/env-nextjs": "0.12.0", + "drizzle-orm": "0.45.2", + "pg": "8.21.0", + "zod": "3.25.76" + }, + "devDependencies": { + "@types/bun": "1.3.14", + "@types/node": "20.19.39", + "@types/pg": "8.18.0", + "typescript": "5.9.3" + } +} diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts new file mode 100644 index 0000000..b689f78 --- /dev/null +++ b/apps/worker/src/index.ts @@ -0,0 +1,128 @@ +import { randomUUID } from "node:crypto"; + +import { generateRecurringInvoice } from "../../web/src/server/jobs/handlers/recurring-invoice"; +import { + claimNextJob, + completeJob, + failJob, + jobTypes, + scheduleDueRecurringInvoiceJobs, + type BackgroundJob, +} from "../../web/src/server/jobs/queue"; + +const workerId = `beenvoice-worker:${randomUUID()}`; +const pollMs = Number(process.env.WORKER_POLL_MS ?? 2_000); +const scheduleMs = Number(process.env.WORKER_SCHEDULE_MS ?? 60_000); +let stopping = false; +let working = false; +let scheduling = false; + +function log(level: "info" | "error", event: string, fields: Record = {}) { + const record = JSON.stringify({ + timestamp: new Date().toISOString(), + level, + service: "beenvoice-worker", + event, + workerId, + ...fields, + }); + if (level === "error") console.error(record); + else console.info(record); +} + +function errorFields(error: unknown) { + if (!(error instanceof Error)) return { error: String(error) }; + const cause = error.cause; + return { + error: error.message || error.name, + ...(cause instanceof Error + ? { errorCause: cause.message || cause.name } + : cause + ? { errorCause: String(cause) } + : {}), + }; +} + +async function scheduleRecurringInvoices() { + if (scheduling) return; + scheduling = true; + try { + const result = await scheduleDueRecurringInvoiceJobs(); + if (result.due) log("info", "scheduler.recurring_invoices", result); + } finally { + scheduling = false; + } +} + +async function handleJob(job: BackgroundJob) { + if (job.type === jobTypes.generateRecurringInvoice) { + await generateRecurringInvoice(job); + return; + } + throw new Error(`No handler registered for ${job.type}`); +} + +async function drainJobs() { + if (working) return; + working = true; + try { + while (!stopping) { + const job = await claimNextJob(workerId); + if (!job) break; + try { + await handleJob(job); + await completeJob(job.id); + log("info", "job.completed", { jobId: job.id, jobType: job.type, attempts: job.attempts }); + } catch (error) { + const terminal = await failJob(job, error); + log("error", "job.failed", { + jobId: job.id, + jobType: job.type, + attempts: job.attempts, + terminal, + ...errorFields(error), + }); + } + } + } finally { + working = false; + } +} + +async function runSchedulerTick() { + try { + await scheduleRecurringInvoices(); + } catch (error) { + log("error", "scheduler.failed", errorFields(error)); + } +} + +async function runWorkerTick() { + try { + await drainJobs(); + } catch (error) { + log("error", "worker.tick_failed", errorFields(error)); + } +} + +function shutdown(signal: string) { + if (stopping) return; + stopping = true; + log("info", "worker.stopping", { signal }); +} + +process.on("SIGTERM", () => shutdown("SIGTERM")); +process.on("SIGINT", () => shutdown("SIGINT")); + +log("info", "worker.started", { pollMs, scheduleMs }); +await runSchedulerTick(); +await runWorkerTick(); + +const scheduleTimer = setInterval(() => void runSchedulerTick(), scheduleMs); +const workTimer = setInterval(() => void runWorkerTick(), pollMs); + +while (!stopping) await Bun.sleep(250); +clearInterval(scheduleTimer); +clearInterval(workTimer); +while (working || scheduling) await Bun.sleep(100); +log("info", "worker.stopped"); diff --git a/apps/worker/tests/recurring-invoices.test.ts b/apps/worker/tests/recurring-invoices.test.ts new file mode 100644 index 0000000..34c983f --- /dev/null +++ b/apps/worker/tests/recurring-invoices.test.ts @@ -0,0 +1,18 @@ +import { describe, expect, test } from "bun:test"; + +import { nextDueDate } from "../../web/src/server/services/recurring-invoices"; + +describe("recurring invoice scheduling", () => { + test("advances weekly schedules from their scheduled occurrence", () => { + const scheduledFor = new Date("2026-08-17T12:00:00.000Z"); + expect(nextDueDate("weekly", scheduledFor).toISOString()).toBe( + "2026-08-24T12:00:00.000Z", + ); + }); + + test("does not mutate the source date", () => { + const scheduledFor = new Date("2026-08-17T12:00:00.000Z"); + nextDueDate("monthly", scheduledFor); + expect(scheduledFor.toISOString()).toBe("2026-08-17T12:00:00.000Z"); + }); +}); diff --git a/apps/worker/tsconfig.json b/apps/worker/tsconfig.json new file mode 100644 index 0000000..7a1203e --- /dev/null +++ b/apps/worker/tsconfig.json @@ -0,0 +1,14 @@ +{ + "extends": "../web/tsconfig.json", + "compilerOptions": { + "baseUrl": ".", + "paths": { + "~/*": ["../web/src/*"], + "src/*": ["../web/src/*"] + }, + "incremental": false, + "jsx": "preserve", + "types": ["node", "bun"] + }, + "include": ["src/**/*.ts", "tests/**/*.ts", "../web/src/server/**/*.ts", "../web/src/env.js"] +} diff --git a/bun.lock b/bun.lock index 485a3f3..5dd52d7 100644 --- a/bun.lock +++ b/bun.lock @@ -168,6 +168,22 @@ "typescript-eslint": "8.60.1", }, }, + "apps/worker": { + "name": "@beenvoice/worker", + "version": "0.1.0", + "dependencies": { + "@t3-oss/env-nextjs": "0.12.0", + "drizzle-orm": "0.45.2", + "pg": "8.21.0", + "zod": "3.25.76", + }, + "devDependencies": { + "@types/bun": "1.3.14", + "@types/node": "20.19.39", + "@types/pg": "8.18.0", + "typescript": "5.9.3", + }, + }, "packages/domain": { "name": "@beenvoice/domain", "version": "0.0.1", @@ -432,6 +448,8 @@ "@beenvoice/web": ["@beenvoice/web@workspace:apps/web"], + "@beenvoice/worker": ["@beenvoice/worker@workspace:apps/worker"], + "@better-auth/core": ["@better-auth/core@1.6.29", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.39.0", "@standard-schema/spec": "^1.1.0", "zod": "^4.3.6" }, "peerDependencies": { "@better-auth/utils": "0.4.2", "@better-fetch/fetch": "1.3.1", "@cloudflare/workers-types": ">=4", "@opentelemetry/api": "^1.9.0", "better-call": "1.4.0", "jose": "^6.1.0", "kysely": "^0.28.5 || ^0.29.0", "nanostores": "^1.0.1" }, "optionalPeers": ["@cloudflare/workers-types", "@opentelemetry/api"] }, "sha512-hkUxePo548G0Y247had36TcfsH7E+cofcTc2UivMAbjkdLPKn1ZUk6Pjkc/kvk9xTbWYCUDX6JA9emJc0Gnufg=="], "@better-auth/drizzle-adapter": ["@better-auth/drizzle-adapter@1.6.19", "", { "peerDependencies": { "@better-auth/core": "^1.6.19", "@better-auth/utils": "0.4.2", "drizzle-orm": "^0.45.2" }, "optionalPeers": ["drizzle-orm"] }, "sha512-57C9ePorPmIEez6dHuQMz3hCTkYim0lfVRIoRtX7PiVfiRFB2bjXseQwrCJfQmkgMFlkp1s/c9nKgAjc2EvAIg=="], @@ -2718,6 +2736,8 @@ "@beenvoice/web/react-dom": ["react-dom@19.2.8", "", { "dependencies": { "scheduler": "^0.27.0" }, "peerDependencies": { "react": "^19.2.8" } }, "sha512-rVprimfGBG3DR+Tq0IQG2DT5PxKth1WIGDmj5yPmlzr4YBe7uyE+Du4oVqTDXZSHGGGXRtTJEGSSePyQCMBglQ=="], + "@beenvoice/worker/@types/pg": ["@types/pg@8.18.0", "", { "dependencies": { "@types/node": "*", "pg-protocol": "*", "pg-types": "^2.2.0" } }, "sha512-gT+oueVQkqnj6ajGJXblFR4iavIXWsGAFCk3dP4Kki5+a9R4NMt0JARdk6s8cUKcfUoqP5dAtDSLU8xYUTFV+Q=="], + "@better-auth/core/better-call": ["better-call@1.4.0", "", { "dependencies": { "@better-auth/utils": "^0.5.0", "@better-fetch/fetch": "^1.3.1", "rou3": "^0.9.1", "set-cookie-parser": "^3.1.2" }, "peerDependencies": { "zod": "^4.0.0" }, "optionalPeers": ["zod"] }, "sha512-bBKOT4vv1kZLDgxVePdilk/Jwkn+dtRRsmi3DzHcDP+WnswyVl6dR59l2HEeP/0cB+bDoopASAesWDPIdd/zZA=="], "@better-auth/core/zod": ["zod@4.4.3", "", {}, "sha512-ytENFjIJFl2UwYglde2jchW2Hwm4GJFLDiSXWdTrJQBIN9Fcyp7n4DhxJEiWNAJMV1/BqWfW/kkg71UDcHJyTQ=="], @@ -2982,6 +3002,8 @@ "@beenvoice/web/better-auth/zod": ["zod@4.4.3", "", {}, "sha512-ytENFjIJFl2UwYglde2jchW2Hwm4GJFLDiSXWdTrJQBIN9Fcyp7n4DhxJEiWNAJMV1/BqWfW/kkg71UDcHJyTQ=="], + "@beenvoice/worker/@types/pg/@types/node": ["@types/node@20.19.43", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-6oYBAi5ikg4Pl+kGsoYtawUMBT2zZMCvPNF7pVLnHZfd1zf38DRiWn/gT01RYCdUqkv7Fhr+C9ot4/tb+2sVvA=="], + "@better-auth/core/better-call/@better-auth/utils": ["@better-auth/utils@0.5.0", "", { "dependencies": { "@noble/hashes": "^2.0.1" } }, "sha512-BL8W4EfIZFwlu0r54m3v1ztjDhu6dDe/amLTm0xybmbZaNgYUqhD3SjpAsnq0q8YD6/ki4iwIgxJNLP/N3TxiA=="], "@better-auth/core/better-call/rou3": ["rou3@0.9.2", "", {}, "sha512-3SOzvaAg8rkHrXtRjpCvCvbyO5to9oOO27Z/XqHEYXfMRVSw/qMIVdmaOk9W2lcRLtR6dlqTjo9hDeJk70QBYQ=="], diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 1e2e942..afa9cab 100644 --- a/docker-compose.coolify.yml +++ b/docker-compose.coolify.yml @@ -51,6 +51,25 @@ services: condition: service_healthy restart: unless-stopped + worker: + build: + context: . + target: worker + image: ${BEENVOICE_WORKER_IMAGE:-beenvoice-worker:coolify} + environment: + NODE_ENV: production + DATABASE_URL: postgres://${POSTGRES_USER:-postgres}:${POSTGRES_PASSWORD:-postgres}@db:5432/${POSTGRES_DB:-postgres} + DB_DISABLE_SSL: "true" + AUTH_SECRET: ${AUTH_SECRET:?Set AUTH_SECRET in Coolify env} + WORKER_POLL_MS: ${WORKER_POLL_MS:-2000} + WORKER_SCHEDULE_MS: ${WORKER_SCHEDULE_MS:-60000} + RESEND_API_KEY: ${RESEND_API_KEY:-} + RESEND_DOMAIN: ${RESEND_DOMAIN:-} + depends_on: + db: + condition: service_healthy + restart: unless-stopped + db: image: postgres:17-alpine environment: diff --git a/docker-compose.yml b/docker-compose.yml index a28a5be..acebe22 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -50,6 +50,25 @@ services: condition: service_healthy restart: unless-stopped + worker: + build: + context: . + target: worker + image: ${BEENVOICE_WORKER_IMAGE:-beenvoice-worker:local} + environment: + NODE_ENV: production + AUTH_SECRET: ${AUTH_SECRET:?Set AUTH_SECRET in .env} + DATABASE_URL: postgres://${POSTGRES_USER:-postgres}:${POSTGRES_PASSWORD:-postgres}@db:5432/${POSTGRES_DB:-postgres} + DB_DISABLE_SSL: "true" + WORKER_POLL_MS: ${WORKER_POLL_MS:-2000} + WORKER_SCHEDULE_MS: ${WORKER_SCHEDULE_MS:-60000} + RESEND_API_KEY: ${RESEND_API_KEY:-} + RESEND_DOMAIN: ${RESEND_DOMAIN:-} + depends_on: + db: + condition: service_healthy + restart: unless-stopped + db: image: postgres:17-alpine environment: diff --git a/scripts/docker-deploy.sh b/scripts/docker-deploy.sh index d68f7f9..845bd03 100755 --- a/scripts/docker-deploy.sh +++ b/scripts/docker-deploy.sh @@ -3,7 +3,7 @@ set -euo pipefail # Production deploy helper for docker-compose.yml (not docker-compose.dev.yml). # Rebuilds the app image from the current working tree, then starts/restarts services -# (app, db, garage). Receipt storage uses in-stack Garage unless S3_* are +# (app, worker, db, garage). Receipt storage uses in-stack Garage unless S3_* are # overridden in .env. Garage S3 API: localhost:${GARAGE_API_PORT:-3900}. # # Plain `docker compose up -d` reuses the local image tag and does NOT pick up @@ -18,15 +18,17 @@ if [[ -f apps/web/.env ]]; then set +a fi -if [[ -z "${BEENVOICE_IMAGE:-}" ]] && command -v git >/dev/null 2>&1; then +if command -v git >/dev/null 2>&1; then if git rev-parse --is-inside-work-tree >/dev/null 2>&1; then - BEENVOICE_IMAGE="beenvoice:$(git rev-parse --short HEAD)" - export BEENVOICE_IMAGE + BEENVOICE_DEPLOY_SHA="$(git rev-parse --short HEAD)" + BEENVOICE_IMAGE="${BEENVOICE_IMAGE:-beenvoice:${BEENVOICE_DEPLOY_SHA}}" + BEENVOICE_WORKER_IMAGE="${BEENVOICE_WORKER_IMAGE:-beenvoice-worker:${BEENVOICE_DEPLOY_SHA}}" fi fi BEENVOICE_IMAGE="${BEENVOICE_IMAGE:-beenvoice:local}" -export BEENVOICE_IMAGE +BEENVOICE_WORKER_IMAGE="${BEENVOICE_WORKER_IMAGE:-beenvoice-worker:local}" +export BEENVOICE_IMAGE BEENVOICE_WORKER_IMAGE -echo "Deploying ${BEENVOICE_IMAGE} (docker compose up -d --build)..." +echo "Deploying ${BEENVOICE_IMAGE} and ${BEENVOICE_WORKER_IMAGE} (docker compose up -d --build)..." exec docker compose up -d --build "$@" diff --git a/turbo.json b/turbo.json index 91493c5..cc3414c 100644 --- a/turbo.json +++ b/turbo.json @@ -15,6 +15,9 @@ "@beenvoice/mobile#build": { "outputs": [] }, + "@beenvoice/worker#build": { + "outputs": [] + }, "typecheck": { "dependsOn": ["^typecheck"], "outputs": []