From a8296b72263a14219a970f957f4c483bd9c78ea8 Mon Sep 17 00:00:00 2001 From: Sherin Thomas Date: Wed, 21 Oct 2026 19:26:36 +0530 Subject: [PATCH 1/4] Add routes for birthday mail service --- .env.example | 3 + .../migration.sql | 33 ++ prisma/schema.prisma | 31 ++ src/controllers/birthdayEmail.controller.ts | 101 +++++ src/middleware/internalAuth.ts | 54 +++ src/routes/index.ts | 3 +- src/routes/internalBirthday.ts | 27 ++ src/services/birthdayEmail.service.ts | 391 +++++++++++++++++ tests/BirthdayEmail.test.ts | 406 ++++++++++++++++++ 9 files changed, 1048 insertions(+), 1 deletion(-) create mode 100644 prisma/migrations/20261021120000_add_birthday_email_log/migration.sql create mode 100644 src/controllers/birthdayEmail.controller.ts create mode 100644 src/middleware/internalAuth.ts create mode 100644 src/routes/internalBirthday.ts create mode 100644 src/services/birthdayEmail.service.ts create mode 100644 tests/BirthdayEmail.test.ts diff --git a/.env.example b/.env.example index 6d8cbae..52efaef 100644 --- a/.env.example +++ b/.env.example @@ -17,3 +17,6 @@ SUPABASE_SERVICE_ROLE_KEY= # Environment variables for the application NODE_ENV=development +# Service-to-service internal authentication secret (used by codenest backend) +INTERNAL_SERVICE_SECRET= + diff --git a/prisma/migrations/20261021120000_add_birthday_email_log/migration.sql b/prisma/migrations/20261021120000_add_birthday_email_log/migration.sql new file mode 100644 index 0000000..640541c --- /dev/null +++ b/prisma/migrations/20261021120000_add_birthday_email_log/migration.sql @@ -0,0 +1,33 @@ +-- CreateEnum +CREATE TYPE "BirthdayEmailStatus" AS ENUM ('PENDING', 'PROCESSING', 'SENT', 'FAILED', 'DEAD'); + +-- CreateTable +CREATE TABLE "BirthdayEmailLog" ( + "id" TEXT NOT NULL, + "memberId" TEXT NOT NULL, + "birthdayYear" INTEGER NOT NULL, + "status" "BirthdayEmailStatus" NOT NULL DEFAULT 'PENDING', + "attemptCount" INTEGER NOT NULL DEFAULT 0, + "nextRetryAt" TIMESTAMP(3), + "lockedUntil" TIMESTAMP(3), + "lockToken" TEXT, + "providerMessageId" TEXT, + "sentAt" TIMESTAMP(3), + "lastError" TEXT, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "BirthdayEmailLog_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "BirthdayEmailLog_memberId_birthdayYear_key" ON "BirthdayEmailLog"("memberId", "birthdayYear"); + +-- CreateIndex +CREATE INDEX "BirthdayEmailLog_status_nextRetryAt_idx" ON "BirthdayEmailLog"("status", "nextRetryAt"); + +-- CreateIndex +CREATE INDEX "BirthdayEmailLog_status_lockedUntil_idx" ON "BirthdayEmailLog"("status", "lockedUntil"); + +-- AddForeignKey +ALTER TABLE "BirthdayEmailLog" ADD CONSTRAINT "BirthdayEmailLog_memberId_fkey" FOREIGN KEY ("memberId") REFERENCES "Member"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 0d277c4..87bc4de 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -62,6 +62,7 @@ model Member { updatedSiteActions SiteAction[] @relation("SiteActionUpdatedBy") createdEmailTemplates EmailTemplate[] @relation("EmailTemplateCreatedBy") updatedEmailTemplates EmailTemplate[] @relation("EmailTemplateUpdatedBy") + birthdayEmailLogs BirthdayEmailLog[] } model Account { @@ -249,3 +250,33 @@ model EmailTemplate { updatedById String? updatedAt DateTime @updatedAt } + +enum BirthdayEmailStatus { + PENDING + PROCESSING + SENT + FAILED + DEAD +} + +model BirthdayEmailLog { + id String @id @default(uuid()) + memberId String + birthdayYear Int + status BirthdayEmailStatus @default(PENDING) + attemptCount Int @default(0) + nextRetryAt DateTime? + lockedUntil DateTime? + lockToken String? + providerMessageId String? + sentAt DateTime? + lastError String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + member Member @relation(fields: [memberId], references: [id], onDelete: Cascade) + + @@unique([memberId, birthdayYear]) + @@index([status, nextRetryAt]) + @@index([status, lockedUntil]) +} diff --git a/src/controllers/birthdayEmail.controller.ts b/src/controllers/birthdayEmail.controller.ts new file mode 100644 index 0000000..d74d330 --- /dev/null +++ b/src/controllers/birthdayEmail.controller.ts @@ -0,0 +1,101 @@ +import { Request, Response } from "express"; +import * as birthdayEmailService from "../services/birthdayEmail.service"; +import { ApiError } from "../utils/apiError"; + +/** + * POST /api/v1/birthday-emails/initialize + */ +export async function initializeJobs(req: Request, res: Response): Promise { + const { birthdayYear, memberIds } = req.body || {}; + + if (typeof birthdayYear !== "number") { + throw new ApiError("birthdayYear is required and must be a number", 400); + } + + if (!Array.isArray(memberIds)) { + throw new ApiError("memberIds is required and must be an array of strings", 400); + } + + const result = await birthdayEmailService.initializeBirthdayJobs( + birthdayYear, + memberIds + ); + + res.status(200).json({ + success: true, + message: "Birthday jobs initialized", + data: result, + }); +} + +/** + * POST /api/v1/birthday-emails/claim + */ +export async function claimJobs(req: Request, res: Response): Promise { + const { batchSize, leaseDurationMs } = req.body || {}; + + const jobs = await birthdayEmailService.claimBirthdayJobs( + typeof batchSize === "number" ? batchSize : 10, + typeof leaseDurationMs === "number" ? leaseDurationMs : 300000 + ); + + res.status(200).json({ + success: true, + count: jobs.length, + jobs, + }); +} + +/** + * POST /api/v1/birthday-emails/:id/sent + */ +export async function markJobSent(req: Request, res: Response): Promise { + const { id } = req.params; + const { lockToken, providerMessageId, sentAt } = req.body || {}; + + if (!lockToken || typeof lockToken !== "string") { + throw new ApiError("lockToken is required in request body", 400); + } + + const result = await birthdayEmailService.recordBirthdayJobSent( + id, + lockToken, + providerMessageId, + sentAt + ); + + res.status(200).json(result); +} + +/** + * POST /api/v1/birthday-emails/:id/failed + */ +export async function markJobFailed(req: Request, res: Response): Promise { + const { id } = req.params; + const { lockToken, errorClassification, errorMessage, isRetryable } = req.body || {}; + + if (!lockToken || typeof lockToken !== "string") { + throw new ApiError("lockToken is required in request body", 400); + } + + const result = await birthdayEmailService.recordBirthdayJobFailed( + id, + lockToken, + errorClassification ?? "TRANSIENT", + errorMessage ?? "Unknown failure", + typeof isRetryable === "boolean" ? isRetryable : true + ); + + res.status(200).json(result); +} + +/** + * GET /api/v1/birthday-emails/stats + */ +export async function getStats(_req: Request, res: Response): Promise { + const stats = await birthdayEmailService.getBirthdayEmailStats(); + res.status(200).json({ + success: true, + stats, + }); +} diff --git a/src/middleware/internalAuth.ts b/src/middleware/internalAuth.ts new file mode 100644 index 0000000..74fee0f --- /dev/null +++ b/src/middleware/internalAuth.ts @@ -0,0 +1,54 @@ +import { Request, Response, NextFunction } from "express"; +import crypto from "crypto"; +import { ApiError } from "../utils/apiError"; + +/** + * Service-to-service authentication middleware for internal APIs. + * Compares incoming secret header against configured environment secret using constant-time comparison. + */ +export function requireInternalAuth( + req: Request, + _res: Response, + next: NextFunction +): void { + const configuredSecret = + process.env.INTERNAL_SERVICE_SECRET || process.env.BIRTHDAY_SERVICE_SECRET; + + if (!configuredSecret) { + throw new ApiError( + "Service authentication is not configured on server", + 500 + ); + } + + const authHeader = req.headers["authorization"]; + let providedSecret = + (req.headers["x-internal-secret"] as string) || + (req.headers["x-service-secret"] as string); + + if (!providedSecret && authHeader && authHeader.startsWith("Bearer ")) { + providedSecret = authHeader.substring(7).trim(); + } + + if (!providedSecret) { + throw new ApiError( + "Unauthorized: Missing internal service secret header", + 401 + ); + } + + const providedBuffer = Buffer.from(providedSecret, "utf-8"); + const expectedBuffer = Buffer.from(configuredSecret, "utf-8"); + + if ( + providedBuffer.length !== expectedBuffer.length || + !crypto.timingSafeEqual(providedBuffer, expectedBuffer) + ) { + throw new ApiError( + "Unauthorized: Invalid internal service secret", + 401 + ); + } + + next(); +} diff --git a/src/routes/index.ts b/src/routes/index.ts index 74ddb5b..791a9fa 100644 --- a/src/routes/index.ts +++ b/src/routes/index.ts @@ -1,6 +1,5 @@ import { Router } from 'express' import { Multer } from 'multer' -import { SupabaseClient } from '@supabase/supabase-js' import projectsRouter from './projects' import acheivementsRouter from './achievements' import interviewRouter from './interviews' @@ -10,6 +9,7 @@ import progressRouter from './progress' import membersRouter from './members' import siteContentRouter from './site-content' import emailRouter from './email' +import internalBirthdayRouter from './internalBirthday' export default function routes(upload: Multer) { const router = Router(); @@ -29,6 +29,7 @@ export default function routes(upload: Multer) { router.use("/site-content", siteContentRouter(upload)); router.use("/email", emailRouter()); + router.use("/birthday-emails", internalBirthdayRouter()); return router; } diff --git a/src/routes/internalBirthday.ts b/src/routes/internalBirthday.ts new file mode 100644 index 0000000..dba0387 --- /dev/null +++ b/src/routes/internalBirthday.ts @@ -0,0 +1,27 @@ +import { Router } from "express"; +import * as birthdayCtrl from "../controllers/birthdayEmail.controller"; +import { requireInternalAuth } from "../middleware/internalAuth"; + +export default function internalBirthdayRouter(): Router { + const router = Router(); + + // Protect all internal birthday tracking endpoints with service-to-service auth + router.use(requireInternalAuth); + + // Initialize jobs idempotently + router.post("/initialize", birthdayCtrl.initializeJobs); + + // Atomically claim a batch of eligible jobs + router.post("/claim", birthdayCtrl.claimJobs); + + // Statistics + router.get("/stats", birthdayCtrl.getStats); + + // Report job send result + router.post("/:id/sent", birthdayCtrl.markJobSent); + + // Report job failure result + router.post("/:id/failed", birthdayCtrl.markJobFailed); + + return router; +} diff --git a/src/services/birthdayEmail.service.ts b/src/services/birthdayEmail.service.ts new file mode 100644 index 0000000..f61b135 --- /dev/null +++ b/src/services/birthdayEmail.service.ts @@ -0,0 +1,391 @@ +import prisma from "../db/client"; +import { ApiError } from "../utils/apiError"; +import crypto from "crypto"; + +export interface ClaimedJob { + id: string; + memberId: string; + birthdayYear: number; + attemptCount: number; + lockToken: string; +} + +export interface JobStats { + pending: number; + processing: number; + sent: number; + failed: number; + dead: number; + dueForRetry: number; + total: number; +} + +/** Retry delays in milliseconds based on attempt count */ +const RETRY_DELAYS_MS: Record = { + 1: 5 * 60 * 1000, // 5 min + 2: 15 * 60 * 1000, // 15 min + 3: 30 * 60 * 1000, // 30 min + 4: 120 * 60 * 1000, // 120 min +}; + +const QUOTA_RETRY_DELAY_MS = 4 * 60 * 60 * 1000; // 4 hours for quota exhaustion +const MAX_ATTEMPTS = 5; + +/** + * Sanitize error message to ensure no secrets, keys, or bearer tokens are stored. + */ +function sanitizeErrorMessage(msg: string): string { + if (!msg || typeof msg !== "string") return "Unknown error"; + return msg + .replace(/(bearer\s+)[^\s]+/gi, "$1[REDACTED]") + .replace(/(api[-_]?key\s*[:=]\s*)[^\s]+/gi, "$1[REDACTED]") + .replace(/(secret\s*[:=]\s*)[^\s]+/gi, "$1[REDACTED]") + .replace(/(re_[a-zA-Z0-9_-]{20,})/g, "[REDACTED_RESEND_KEY]") + .substring(0, 500); +} + +/** + * Initialize birthday email jobs for a given year and list of member IDs idempotently. + */ +export async function initializeBirthdayJobs( + birthdayYear: number, + memberIds: string[] +): Promise<{ + created: number; + existing: number; + totalEligible: number; +}> { + if (!Number.isInteger(birthdayYear) || birthdayYear < 1900 || birthdayYear > 2200) { + throw new ApiError("Invalid birthdayYear: must be a 4-digit integer", 400); + } + + if (!Array.isArray(memberIds) || memberIds.length === 0) { + throw new ApiError("memberIds must be a non-empty array of strings", 400); + } + + const cleanIds = Array.from( + new Set(memberIds.filter((id) => typeof id === "string" && id.trim().length > 0)) + ); + + if (cleanIds.length === 0) { + throw new ApiError("No valid member IDs provided", 400); + } + + // Fetch only eligible members: approved, not ghosted, has birth_date, has valid email + const eligibleMembers = await prisma.member.findMany({ + where: { + id: { in: cleanIds }, + isApproved: true, + isGhosted: false, + birth_date: { not: null }, + }, + select: { + id: true, + email: true, + }, + }); + + const validEligible = eligibleMembers.filter( + (m) => typeof m.email === "string" && m.email.includes("@") + ); + + if (validEligible.length === 0) { + return { created: 0, existing: 0, totalEligible: 0 }; + } + + const eligibleIds = validEligible.map((m) => m.id); + + // Check existing logs to preserve state and avoid resetting SENT or attempt counts + const existingLogs = await prisma.birthdayEmailLog.findMany({ + where: { + birthdayYear, + memberId: { in: eligibleIds }, + }, + select: { + memberId: true, + }, + }); + + const existingMemberIdSet = new Set(existingLogs.map((l) => l.memberId)); + const newMemberIds = eligibleIds.filter((id) => !existingMemberIdSet.has(id)); + + let createdCount = 0; + if (newMemberIds.length > 0) { + const recordsToCreate = newMemberIds.map((mId) => ({ + memberId: mId, + birthdayYear, + status: "PENDING" as const, + attemptCount: 0, + })); + + const result = await prisma.birthdayEmailLog.createMany({ + data: recordsToCreate, + skipDuplicates: true, + }); + createdCount = result.count; + } + + return { + created: createdCount, + existing: existingLogs.length, + totalEligible: validEligible.length, + }; +} + +/** + * Atomically claim a batch of eligible jobs for processing. + * A job is claimable when: + * - PENDING + * - FAILED and nextRetryAt <= now + * - PROCESSING and lease expired (lockedUntil < now) + * Prevents race conditions with FOR UPDATE SKIP LOCKED. + */ +export async function claimBirthdayJobs( + batchSize = 10, + leaseDurationMs = 300000 +): Promise { + const safeBatchSize = Math.max(1, Math.min(batchSize, 50)); + const safeLeaseMs = Math.max(10000, Math.min(leaseDurationMs, 3600000)); + const now = new Date(); + + return await prisma.$transaction(async (tx) => { + // Select eligible jobs with row-level locks, skipping already-locked rows + const eligible = await tx.$queryRaw>` + SELECT id FROM "BirthdayEmailLog" + WHERE ( + "status" = 'PENDING'::"BirthdayEmailStatus" OR + ("status" = 'FAILED'::"BirthdayEmailStatus" AND ("nextRetryAt" IS NULL OR "nextRetryAt" <= ${now})) OR + ("status" = 'PROCESSING'::"BirthdayEmailStatus" AND "lockedUntil" < ${now}) + ) + ORDER BY "createdAt" ASC + LIMIT ${safeBatchSize} + FOR UPDATE SKIP LOCKED + `; + + if (!eligible || eligible.length === 0) { + return []; + } + + const lockedUntil = new Date(now.getTime() + safeLeaseMs); + const claimed: ClaimedJob[] = []; + + for (const row of eligible) { + const lockToken = crypto.randomUUID(); + const updated = await tx.birthdayEmailLog.update({ + where: { id: row.id }, + data: { + status: "PROCESSING", + lockToken, + lockedUntil, + }, + select: { + id: true, + memberId: true, + birthdayYear: true, + attemptCount: true, + lockToken: true, + }, + }); + + if (updated.lockToken) { + claimed.push({ + id: updated.id, + memberId: updated.memberId, + birthdayYear: updated.birthdayYear, + attemptCount: updated.attemptCount, + lockToken: updated.lockToken, + }); + } + } + + return claimed; + }); +} + +/** + * Mark a claimed birthday email job as SENT. + * Validates the lockToken to guard against stale workers. + */ +export async function recordBirthdayJobSent( + id: string, + lockToken: string, + providerMessageId?: string, + sentAt?: string +): Promise<{ success: boolean; message: string }> { + if (!id || typeof id !== "string") { + throw new ApiError("Missing tracking record id", 400); + } + if (!lockToken || typeof lockToken !== "string") { + throw new ApiError("Missing lockToken", 400); + } + + const record = await prisma.birthdayEmailLog.findUnique({ + where: { id }, + }); + + if (!record) { + throw new ApiError("Birthday email record not found", 404); + } + + // Idempotent duplicate check: already SENT with provider ID + if (record.status === "SENT") { + return { success: true, message: "Record was already marked as sent" }; + } + + // Validate lease ownership + if (record.status !== "PROCESSING" || record.lockToken !== lockToken) { + throw new ApiError( + "Invalid or expired lock token: job is no longer owned by caller", + 409 + ); + } + + await prisma.birthdayEmailLog.update({ + where: { id }, + data: { + status: "SENT", + providerMessageId: providerMessageId ?? record.providerMessageId, + sentAt: sentAt ? new Date(sentAt) : new Date(), + lockToken: null, + lockedUntil: null, + nextRetryAt: null, + attemptCount: record.attemptCount + 1, + lastError: null, + }, + }); + + return { success: true, message: "Marked as sent" }; +} + +/** + * Record a failed attempt for a claimed job. + * Calculates next retry time or transitions to DEAD if max attempts exceeded or permanent failure. + */ +export async function recordBirthdayJobFailed( + id: string, + lockToken: string, + errorClassification: string, + errorMessage: string, + isRetryable: boolean +): Promise<{ + success: boolean; + status: "FAILED" | "DEAD"; + attemptCount: number; + nextRetryAt: Date | null; +}> { + if (!id || typeof id !== "string") { + throw new ApiError("Missing tracking record id", 400); + } + if (!lockToken || typeof lockToken !== "string") { + throw new ApiError("Missing lockToken", 400); + } + + const record = await prisma.birthdayEmailLog.findUnique({ + where: { id }, + }); + + if (!record) { + throw new ApiError("Birthday email record not found", 404); + } + + if (record.status !== "PROCESSING" || record.lockToken !== lockToken) { + throw new ApiError( + "Invalid or expired lock token: job is no longer owned by caller", + 409 + ); + } + + const newAttemptCount = record.attemptCount + 1; + const sanitized = sanitizeErrorMessage(errorMessage); + + const isPermanent = + !isRetryable || + errorClassification === "PERMANENT" || + errorClassification === "CONFIG"; + + let newStatus: "FAILED" | "DEAD"; + let nextRetryAt: Date | null = null; + + if (isPermanent || newAttemptCount >= MAX_ATTEMPTS) { + newStatus = "DEAD"; + nextRetryAt = null; + } else { + newStatus = "FAILED"; + let delayMs = RETRY_DELAYS_MS[newAttemptCount] || 120 * 60 * 1000; + if (errorClassification === "QUOTA") { + delayMs = QUOTA_RETRY_DELAY_MS; + } + nextRetryAt = new Date(Date.now() + delayMs); + } + + await prisma.birthdayEmailLog.update({ + where: { id }, + data: { + status: newStatus, + attemptCount: newAttemptCount, + nextRetryAt, + lastError: sanitized, + lockToken: null, + lockedUntil: null, + }, + }); + + return { + success: true, + status: newStatus, + attemptCount: newAttemptCount, + nextRetryAt, + }; +} + +/** + * Fetch tracking statistics without exposing member emails. + */ +export async function getBirthdayEmailStats(): Promise { + const countsByStatus = await prisma.birthdayEmailLog.groupBy({ + by: ["status"], + _count: { id: true }, + }); + + const stats: JobStats = { + pending: 0, + processing: 0, + sent: 0, + failed: 0, + dead: 0, + dueForRetry: 0, + total: 0, + }; + + for (const group of countsByStatus) { + const count = group._count.id; + stats.total += count; + switch (group.status) { + case "PENDING": + stats.pending = count; + break; + case "PROCESSING": + stats.processing = count; + break; + case "SENT": + stats.sent = count; + break; + case "FAILED": + stats.failed = count; + break; + case "DEAD": + stats.dead = count; + break; + } + } + + const dueForRetry = await prisma.birthdayEmailLog.count({ + where: { + status: "FAILED", + nextRetryAt: { lte: new Date() }, + }, + }); + stats.dueForRetry = dueForRetry; + + return stats; +} diff --git a/tests/BirthdayEmail.test.ts b/tests/BirthdayEmail.test.ts new file mode 100644 index 0000000..7e5626a --- /dev/null +++ b/tests/BirthdayEmail.test.ts @@ -0,0 +1,406 @@ +import { Request, Response, NextFunction } from "express"; +import * as birthdayService from "../src/services/birthdayEmail.service"; +import * as birthdayCtrl from "../src/controllers/birthdayEmail.controller"; +import { requireInternalAuth } from "../src/middleware/internalAuth"; +import prisma from "../src/db/client"; +import { ApiError } from "../src/utils/apiError"; + +jest.mock("../src/db/client", () => { + const mockPrisma = { + member: { + findMany: jest.fn(), + findUnique: jest.fn(), + }, + birthdayEmailLog: { + findMany: jest.fn(), + findUnique: jest.fn(), + createMany: jest.fn(), + update: jest.fn(), + groupBy: jest.fn(), + count: jest.fn(), + }, + $transaction: jest.fn(), + $queryRaw: jest.fn(), + }; + return { + __esModule: true, + default: mockPrisma, + prisma: mockPrisma, + }; +}); + +describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { + const originalEnv = process.env; + + beforeEach(() => { + jest.clearAllMocks(); + process.env = { ...originalEnv }; + process.env.INTERNAL_SERVICE_SECRET = "super-secret-service-token"; + }); + + afterAll(() => { + process.env = originalEnv; + }); + + // ────────────────────────────────────────────── + // 1. Initializing the same member/year job twice creates one record + // ────────────────────────────────────────────── + it("1. Initializing the same member/year job twice creates one record", async () => { + const memberId = "member-123"; + (prisma.member.findMany as jest.Mock).mockResolvedValue([ + { id: memberId, email: "alice@example.com" }, + ]); + + // First initialization: no existing log + (prisma.birthdayEmailLog.findMany as jest.Mock).mockResolvedValueOnce([]); + (prisma.birthdayEmailLog.createMany as jest.Mock).mockResolvedValueOnce({ count: 1 }); + + const firstRun = await birthdayService.initializeBirthdayJobs(2026, [memberId]); + expect(firstRun.created).toBe(1); + expect(firstRun.existing).toBe(0); + + // Second initialization: existing log found + (prisma.birthdayEmailLog.findMany as jest.Mock).mockResolvedValueOnce([ + { memberId, status: "PENDING" }, + ]); + + const secondRun = await birthdayService.initializeBirthdayJobs(2026, [memberId]); + expect(secondRun.created).toBe(0); + expect(secondRun.existing).toBe(1); + expect(prisma.birthdayEmailLog.createMany).toHaveBeenCalledTimes(1); + }); + + // ────────────────────────────────────────────── + // 2. Existing SENT records are not reset + // ────────────────────────────────────────────── + it("2. Existing SENT records are not reset during initialization", async () => { + const memberId = "member-sent"; + (prisma.member.findMany as jest.Mock).mockResolvedValue([ + { id: memberId, email: "sent@example.com" }, + ]); + + // Existing record is SENT + (prisma.birthdayEmailLog.findMany as jest.Mock).mockResolvedValueOnce([ + { memberId, status: "SENT", attemptCount: 1 }, + ]); + + const res = await birthdayService.initializeBirthdayJobs(2026, [memberId]); + expect(res.created).toBe(0); + expect(res.existing).toBe(1); + expect(prisma.birthdayEmailLog.update).not.toHaveBeenCalled(); + expect(prisma.birthdayEmailLog.createMany).not.toHaveBeenCalled(); + }); + + // ────────────────────────────────────────────── + // 3. Two concurrent workers cannot claim the same active job + // ────────────────────────────────────────────── + it("3. Atomic job claim locks jobs and prevents concurrent workers from claiming the same active job", async () => { + const jobRow = { id: "job-1" }; + + // Transaction executes queryRaw and update + (prisma.$transaction as jest.Mock).mockImplementation(async (callback) => { + const tx = { + $queryRaw: jest.fn().mockResolvedValue([jobRow]), + birthdayEmailLog: { + update: jest.fn().mockResolvedValue({ + id: "job-1", + memberId: "m-1", + birthdayYear: 2026, + attemptCount: 0, + lockToken: "token-worker-1", + }), + }, + }; + return await callback(tx); + }); + + const claimed = await birthdayService.claimBirthdayJobs(10, 300000); + expect(claimed).toHaveLength(1); + expect(claimed[0].id).toBe("job-1"); + expect(claimed[0].lockToken).toBeDefined(); + + // Second worker claims concurrently, queryRaw returns empty (already locked/skipped) + (prisma.$transaction as jest.Mock).mockImplementationOnce(async (callback) => { + const tx = { + $queryRaw: jest.fn().mockResolvedValue([]), + birthdayEmailLog: { update: jest.fn() }, + }; + return await callback(tx); + }); + + const secondClaim = await birthdayService.claimBirthdayJobs(10, 300000); + expect(secondClaim).toHaveLength(0); + }); + + // ────────────────────────────────────────────── + // 4. Expired PROCESSING jobs can be recovered + // ────────────────────────────────────────────── + it("4. Expired PROCESSING jobs can be recovered by claim query", async () => { + (prisma.$transaction as jest.Mock).mockImplementation(async (callback) => { + const tx = { + $queryRaw: jest.fn().mockImplementation((strings, ...values) => { + // Verify query includes lease expiration check: lockedUntil < now + const queryText = strings.join(""); + expect(queryText).toContain('"lockedUntil" <'); + return Promise.resolve([{ id: "expired-job-1" }]); + }), + birthdayEmailLog: { + update: jest.fn().mockResolvedValue({ + id: "expired-job-1", + memberId: "m-expired", + birthdayYear: 2026, + attemptCount: 1, + lockToken: "new-recovery-token", + }), + }, + }; + return await callback(tx); + }); + + const claimed = await birthdayService.claimBirthdayJobs(5, 300000); + expect(claimed).toHaveLength(1); + expect(claimed[0].id).toBe("expired-job-1"); + expect(claimed[0].lockToken).toBe("new-recovery-token"); + }); + + // ────────────────────────────────────────────── + // 5. A stale lock token cannot overwrite a newer worker's result + // ────────────────────────────────────────────── + it("5. A stale lock token cannot overwrite a newer worker's result (409 conflict)", async () => { + // Record in DB currently has 'newer-lock-token' + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-conflict", + status: "PROCESSING", + lockToken: "newer-lock-token", + attemptCount: 2, + }); + + // Stale worker tries to report sent with old token + await expect( + birthdayService.recordBirthdayJobSent( + "job-conflict", + "stale-old-token", + "msg-123" + ) + ).rejects.toThrow(ApiError); + + try { + await birthdayService.recordBirthdayJobSent( + "job-conflict", + "stale-old-token", + "msg-123" + ); + } catch (err: any) { + expect(err.statusCode).toBe(409); + } + + // Stale worker tries to report failed with old token + await expect( + birthdayService.recordBirthdayJobFailed( + "job-conflict", + "stale-old-token", + "TRANSIENT", + "network timeout", + true + ) + ).rejects.toThrow(ApiError); + }); + + // ────────────────────────────────────────────── + // 6. Successful sends persist provider IDs and timestamps + // ────────────────────────────────────────────── + it("6. Successful sends persist provider IDs and timestamps", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-success", + status: "PROCESSING", + lockToken: "valid-token-abc", + attemptCount: 0, + providerMessageId: null, + }); + + (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({ + id: "job-success", + status: "SENT", + providerMessageId: "resend-msg-999", + sentAt: new Date("2026-10-21T09:05:00.000Z"), + }); + + const res = await birthdayService.recordBirthdayJobSent( + "job-success", + "valid-token-abc", + "resend-msg-999", + "2026-10-21T09:05:00.000Z" + ); + + expect(res.success).toBe(true); + expect(prisma.birthdayEmailLog.update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: "job-success" }, + data: expect.objectContaining({ + status: "SENT", + providerMessageId: "resend-msg-999", + sentAt: new Date("2026-10-21T09:05:00.000Z"), + lockToken: null, + lockedUntil: null, + attemptCount: 1, + }), + }) + ); + }); + + // ────────────────────────────────────────────── + // 7. Retryable failures schedule a future retry + // ────────────────────────────────────────────── + it("7. Retryable failures schedule a future retry with backoff", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-retryable", + status: "PROCESSING", + lockToken: "tok-retry", + attemptCount: 1, // attempt 1 failed, next will be attempt 2 + }); + + (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({}); + + const res = await birthdayService.recordBirthdayJobFailed( + "job-retryable", + "tok-retry", + "TRANSIENT", + "Temporary 500 error", + true + ); + + expect(res.success).toBe(true); + expect(res.status).toBe("FAILED"); + expect(res.attemptCount).toBe(2); + expect(res.nextRetryAt).toBeInstanceOf(Date); + // Attempt 2 backoff is 15 minutes (~900000ms from now) + const diffMs = res.nextRetryAt!.getTime() - Date.now(); + expect(diffMs).toBeGreaterThan(14 * 60 * 1000); + expect(diffMs).toBeLessThanOrEqual(16 * 60 * 1000); + }); + + // ────────────────────────────────────────────── + // 8. Permanent failures become terminal (DEAD) + // ────────────────────────────────────────────── + it("8. Permanent failures become terminal (DEAD) without scheduling retry", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-perm", + status: "PROCESSING", + lockToken: "tok-perm", + attemptCount: 0, + }); + + (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({}); + + const res = await birthdayService.recordBirthdayJobFailed( + "job-perm", + "tok-perm", + "PERMANENT", + "Invalid recipient mailbox not found", + false + ); + + expect(res.success).toBe(true); + expect(res.status).toBe("DEAD"); + expect(res.nextRetryAt).toBeNull(); + expect(res.attemptCount).toBe(1); + expect(prisma.birthdayEmailLog.update).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + status: "DEAD", + nextRetryAt: null, + }), + }) + ); + }); + + // ────────────────────────────────────────────── + // 9. Maximum attempts are enforced + // ────────────────────────────────────────────── + it("9. Maximum attempts (5) are enforced, moving record to DEAD", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-max-attempt", + status: "PROCESSING", + lockToken: "tok-max", + attemptCount: 4, // 5th attempt failing + }); + + (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({}); + + const res = await birthdayService.recordBirthdayJobFailed( + "job-max-attempt", + "tok-max", + "TRANSIENT", + "Server error 503", + true + ); + + expect(res.success).toBe(true); + expect(res.status).toBe("DEAD"); + expect(res.attemptCount).toBe(5); + expect(res.nextRetryAt).toBeNull(); + }); + + // ────────────────────────────────────────────── + // 10. Unauthorized internal API requests are rejected + // ────────────────────────────────────────────── + it("10. Unauthorized internal API requests are rejected by middleware", () => { + const next: NextFunction = jest.fn(); + + // 10a: Missing secret + const reqMissing = { + headers: {}, + } as unknown as Request; + const res = {} as Response; + + expect(() => requireInternalAuth(reqMissing, res, next)).toThrow(ApiError); + try { + requireInternalAuth(reqMissing, res, next); + } catch (err: any) { + expect(err.statusCode).toBe(401); + } + + // 10b: Invalid secret + const reqInvalid = { + headers: { + "x-service-secret": "wrong-secret-token", + }, + } as unknown as Request; + + expect(() => requireInternalAuth(reqInvalid, res, next)).toThrow(ApiError); + try { + requireInternalAuth(reqInvalid, res, next); + } catch (err: any) { + expect(err.statusCode).toBe(401); + } + + // 10c: Valid secret + const reqValid = { + headers: { + "x-service-secret": "super-secret-service-token", + }, + } as unknown as Request; + + expect(() => requireInternalAuth(reqValid, res, next)).not.toThrow(); + expect(next).toHaveBeenCalled(); + }); + + // ────────────────────────────────────────────── + // Stats endpoint + // ────────────────────────────────────────────── + it("Fetches internal stats safely", async () => { + (prisma.birthdayEmailLog.groupBy as jest.Mock).mockResolvedValue([ + { status: "SENT", _count: { id: 10 } }, + { status: "PENDING", _count: { id: 2 } }, + { status: "FAILED", _count: { id: 1 } }, + ]); + (prisma.birthdayEmailLog.count as jest.Mock).mockResolvedValue(1); + + const stats = await birthdayService.getBirthdayEmailStats(); + expect(stats.sent).toBe(10); + expect(stats.pending).toBe(2); + expect(stats.failed).toBe(1); + expect(stats.dueForRetry).toBe(1); + expect(stats.total).toBe(13); + }); +}); From 8cca9d01dd495cdcd67430a303f252a918a77a25 Mon Sep 17 00:00:00 2001 From: Sarvesh Shahane Date: Wed, 21 Oct 2026 20:54:08 +0530 Subject: [PATCH 2/4] fix: address review feedback on birthday email API - Write job results with a lease-checked updateMany so a worker whose lease expired can't overwrite a job another worker has claimed - Count the lost attempt when reclaiming an expired PROCESSING job and move it to DEAD at max attempts, so crashing jobs don't retry forever - Validate sentAt and errorClassification in the controller (400 not 500) - Controller builds the { success: true, ... } response; service returns data - Read INTERNAL_SERVICE_SECRET via config, drop BIRTHDAY_SERVICE_SECRET fallback, hash before timingSafeEqual to hide secret length - Rename route file to birthdayEmails.ts and add apidoc blocks - Dedupe lease checks, defaults and retry delays; type errorClassification - Update and extend tests --- src/config/index.ts | 1 + src/controllers/birthdayEmail.controller.ts | 78 ++-- src/middleware/internalAuth.ts | 20 +- src/routes/birthdayEmails.ts | 122 ++++++ src/routes/index.ts | 4 +- src/routes/internalBirthday.ts | 27 -- src/services/birthdayEmail.service.ts | 242 ++++++------ tests/BirthdayEmail.test.ts | 388 +++++++++++++------- 8 files changed, 575 insertions(+), 307 deletions(-) create mode 100644 src/routes/birthdayEmails.ts delete mode 100644 src/routes/internalBirthday.ts diff --git a/src/config/index.ts b/src/config/index.ts index cc31432..03d7399 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -5,4 +5,5 @@ export default { SUPABASE_URL: process.env.SUPABASE_URL!, SUPABASE_SERVICE_ROLE_KEY: process.env.SUPABASE_SERVICE_ROLE_KEY!, ALLOWED_ORIGINS: process.env.ALLOWED_ORIGINS, + INTERNAL_SERVICE_SECRET: process.env.INTERNAL_SERVICE_SECRET, }; diff --git a/src/controllers/birthdayEmail.controller.ts b/src/controllers/birthdayEmail.controller.ts index d74d330..fdbf7b4 100644 --- a/src/controllers/birthdayEmail.controller.ts +++ b/src/controllers/birthdayEmail.controller.ts @@ -1,7 +1,39 @@ import { Request, Response } from "express"; import * as birthdayEmailService from "../services/birthdayEmail.service"; +import { + ERROR_CLASSIFICATIONS, + ErrorClassification, +} from "../services/birthdayEmail.service"; import { ApiError } from "../utils/apiError"; +function requireLockToken(body: Record): string { + const { lockToken } = body; + if (!lockToken || typeof lockToken !== "string") { + throw new ApiError("lockToken is required in request body", 400); + } + return lockToken; +} + +function parseSentAt(value: unknown): Date | undefined { + if (value === undefined || value === null) return undefined; + const date = new Date(value as string); + if (typeof value !== "string" || Number.isNaN(date.getTime())) { + throw new ApiError("sentAt must be a valid ISO date string", 400); + } + return date; +} + +function parseErrorClassification(value: unknown): ErrorClassification { + if (value === undefined || value === null) return "TRANSIENT"; + if (!ERROR_CLASSIFICATIONS.includes(value as ErrorClassification)) { + throw new ApiError( + `errorClassification must be one of: ${ERROR_CLASSIFICATIONS.join(", ")}`, + 400 + ); + } + return value as ErrorClassification; +} + /** * POST /api/v1/birthday-emails/initialize */ @@ -35,8 +67,8 @@ export async function claimJobs(req: Request, res: Response): Promise { const { batchSize, leaseDurationMs } = req.body || {}; const jobs = await birthdayEmailService.claimBirthdayJobs( - typeof batchSize === "number" ? batchSize : 10, - typeof leaseDurationMs === "number" ? leaseDurationMs : 300000 + typeof batchSize === "number" ? batchSize : undefined, + typeof leaseDurationMs === "number" ? leaseDurationMs : undefined ); res.status(200).json({ @@ -50,43 +82,43 @@ export async function claimJobs(req: Request, res: Response): Promise { * POST /api/v1/birthday-emails/:id/sent */ export async function markJobSent(req: Request, res: Response): Promise { - const { id } = req.params; - const { lockToken, providerMessageId, sentAt } = req.body || {}; + const body = req.body || {}; + const lockToken = requireLockToken(body); + const sentAt = parseSentAt(body.sentAt); - if (!lockToken || typeof lockToken !== "string") { - throw new ApiError("lockToken is required in request body", 400); - } - - const result = await birthdayEmailService.recordBirthdayJobSent( - id, + const { alreadySent } = await birthdayEmailService.recordBirthdayJobSent( + req.params.id, lockToken, - providerMessageId, + body.providerMessageId, sentAt ); - res.status(200).json(result); + res.status(200).json({ + success: true, + message: alreadySent ? "Record was already marked as sent" : "Marked as sent", + }); } /** * POST /api/v1/birthday-emails/:id/failed */ export async function markJobFailed(req: Request, res: Response): Promise { - const { id } = req.params; - const { lockToken, errorClassification, errorMessage, isRetryable } = req.body || {}; - - if (!lockToken || typeof lockToken !== "string") { - throw new ApiError("lockToken is required in request body", 400); - } + const body = req.body || {}; + const lockToken = requireLockToken(body); + const errorClassification = parseErrorClassification(body.errorClassification); const result = await birthdayEmailService.recordBirthdayJobFailed( - id, + req.params.id, lockToken, - errorClassification ?? "TRANSIENT", - errorMessage ?? "Unknown failure", - typeof isRetryable === "boolean" ? isRetryable : true + errorClassification, + typeof body.errorMessage === "string" ? body.errorMessage : "Unknown failure", + typeof body.isRetryable === "boolean" ? body.isRetryable : true ); - res.status(200).json(result); + res.status(200).json({ + success: true, + ...result, + }); } /** diff --git a/src/middleware/internalAuth.ts b/src/middleware/internalAuth.ts index 74fee0f..ce3e282 100644 --- a/src/middleware/internalAuth.ts +++ b/src/middleware/internalAuth.ts @@ -1,18 +1,22 @@ import { Request, Response, NextFunction } from "express"; import crypto from "crypto"; +import config from "../config"; import { ApiError } from "../utils/apiError"; +function sha256(value: string): Buffer { + return crypto.createHash("sha256").update(value, "utf-8").digest(); +} + /** * Service-to-service authentication middleware for internal APIs. - * Compares incoming secret header against configured environment secret using constant-time comparison. + * Compares the incoming secret header against INTERNAL_SERVICE_SECRET in constant time. */ export function requireInternalAuth( req: Request, _res: Response, next: NextFunction ): void { - const configuredSecret = - process.env.INTERNAL_SERVICE_SECRET || process.env.BIRTHDAY_SERVICE_SECRET; + const configuredSecret = config.INTERNAL_SERVICE_SECRET; if (!configuredSecret) { throw new ApiError( @@ -37,13 +41,9 @@ export function requireInternalAuth( ); } - const providedBuffer = Buffer.from(providedSecret, "utf-8"); - const expectedBuffer = Buffer.from(configuredSecret, "utf-8"); - - if ( - providedBuffer.length !== expectedBuffer.length || - !crypto.timingSafeEqual(providedBuffer, expectedBuffer) - ) { + // Hash both sides so the buffers are always equal length and the + // comparison doesn't reveal the secret's length. + if (!crypto.timingSafeEqual(sha256(providedSecret), sha256(configuredSecret))) { throw new ApiError( "Unauthorized: Invalid internal service secret", 401 diff --git a/src/routes/birthdayEmails.ts b/src/routes/birthdayEmails.ts new file mode 100644 index 0000000..cf8d13a --- /dev/null +++ b/src/routes/birthdayEmails.ts @@ -0,0 +1,122 @@ +import { Router } from "express"; +import * as birthdayCtrl from "../controllers/birthdayEmail.controller"; +import { requireInternalAuth } from "../middleware/internalAuth"; + +/** + * Internal, service-to-service API used by the birthday email worker. + * Every route requires the INTERNAL_SERVICE_SECRET, sent as an + * `x-internal-secret` / `x-service-secret` header or as a Bearer token. + */ +export default function birthdayEmailsRouter(): Router { + const router = Router(); + + router.use(requireInternalAuth); + + /** + * @api {post} /birthday-emails/initialize Create birthday email jobs + * @apiName InitializeBirthdayJobs + * @apiGroup BirthdayEmail + * @apiDescription Idempotently creates one PENDING job per eligible member + * (approved, not ghosted, has a birth date and email) for the given year. + * Existing jobs are left untouched. + * + * @apiHeader {String} x-internal-secret Internal service secret. + * + * @apiBody {Number} birthdayYear Year the birthday email is for. (Required) + * @apiBody {String[]} memberIds Member IDs to create jobs for. (Required) + * + * @apiSuccess {Boolean} success Always true. + * @apiSuccess {Object} data + * @apiSuccess {Number} data.created Jobs newly created. + * @apiSuccess {Number} data.existing Jobs that already existed. + * @apiSuccess {Number} data.totalEligible Eligible members found. + * @apiError (400) BadRequest Invalid birthdayYear or memberIds. + * @apiError (401) Unauthorized Missing or invalid service secret. + */ + router.post("/initialize", birthdayCtrl.initializeJobs); + + /** + * @api {post} /birthday-emails/claim Claim a batch of jobs + * @apiName ClaimBirthdayJobs + * @apiGroup BirthdayEmail + * @apiDescription Atomically leases PENDING jobs, FAILED jobs due for retry, + * and PROCESSING jobs whose lease expired. Each claimed job gets a fresh + * lockToken that must be sent back when reporting the result. + * + * @apiHeader {String} x-internal-secret Internal service secret. + * + * @apiBody {Number} [batchSize=10] Jobs to claim (1-50). + * @apiBody {Number} [leaseDurationMs=300000] Lease length in ms (10s-1h). + * + * @apiSuccess {Boolean} success Always true. + * @apiSuccess {Number} count Number of jobs claimed. + * @apiSuccess {Object[]} jobs Claimed jobs (id, memberId, birthdayYear, attemptCount, lockToken). + * @apiError (401) Unauthorized Missing or invalid service secret. + */ + router.post("/claim", birthdayCtrl.claimJobs); + + /** + * @api {get} /birthday-emails/stats Get job statistics + * @apiName GetBirthdayEmailStats + * @apiGroup BirthdayEmail + * + * @apiHeader {String} x-internal-secret Internal service secret. + * + * @apiSuccess {Boolean} success Always true. + * @apiSuccess {Object} stats Counts: pending, processing, sent, failed, dead, dueForRetry, total. + * @apiError (401) Unauthorized Missing or invalid service secret. + */ + router.get("/stats", birthdayCtrl.getStats); + + /** + * @api {post} /birthday-emails/:id/sent Report a sent email + * @apiName MarkBirthdayJobSent + * @apiGroup BirthdayEmail + * @apiDescription Marks a claimed job as SENT. Calling it again for a job + * that is already SENT succeeds without changing anything. + * + * @apiHeader {String} x-internal-secret Internal service secret. + * + * @apiParam (URL Params) {String} id Job ID. + * + * @apiBody {String} lockToken Token returned by /claim. (Required) + * @apiBody {String} [providerMessageId] Email provider's message ID. + * @apiBody {String} [sentAt] ISO timestamp of the send. Defaults to now. + * + * @apiSuccess {Boolean} success Always true. + * @apiSuccess {String} message Result description. + * @apiError (400) BadRequest Missing lockToken or invalid sentAt. + * @apiError (404) NotFound Job not found. + * @apiError (409) Conflict Lease expired or held by another worker. + */ + router.post("/:id/sent", birthdayCtrl.markJobSent); + + /** + * @api {post} /birthday-emails/:id/failed Report a failed send + * @apiName MarkBirthdayJobFailed + * @apiGroup BirthdayEmail + * @apiDescription Records a failed attempt. Retryable failures are + * rescheduled with backoff; permanent failures, or a job's 5th attempt, + * move it to DEAD. + * + * @apiHeader {String} x-internal-secret Internal service secret. + * + * @apiParam (URL Params) {String} id Job ID. + * + * @apiBody {String} lockToken Token returned by /claim. (Required) + * @apiBody {String="TRANSIENT","PERMANENT","CONFIG","QUOTA"} [errorClassification=TRANSIENT] Failure type. + * @apiBody {String} [errorMessage] Error details. Secrets are redacted before storing. + * @apiBody {Boolean} [isRetryable=true] Whether the failure can be retried. + * + * @apiSuccess {Boolean} success Always true. + * @apiSuccess {String} status FAILED or DEAD. + * @apiSuccess {Number} attemptCount Attempts made so far. + * @apiSuccess {String} nextRetryAt Next retry time, or null when DEAD. + * @apiError (400) BadRequest Missing lockToken or invalid errorClassification. + * @apiError (404) NotFound Job not found. + * @apiError (409) Conflict Lease expired or held by another worker. + */ + router.post("/:id/failed", birthdayCtrl.markJobFailed); + + return router; +} diff --git a/src/routes/index.ts b/src/routes/index.ts index 791a9fa..673ff79 100644 --- a/src/routes/index.ts +++ b/src/routes/index.ts @@ -9,7 +9,7 @@ import progressRouter from './progress' import membersRouter from './members' import siteContentRouter from './site-content' import emailRouter from './email' -import internalBirthdayRouter from './internalBirthday' +import birthdayEmailsRouter from './birthdayEmails' export default function routes(upload: Multer) { const router = Router(); @@ -29,7 +29,7 @@ export default function routes(upload: Multer) { router.use("/site-content", siteContentRouter(upload)); router.use("/email", emailRouter()); - router.use("/birthday-emails", internalBirthdayRouter()); + router.use("/birthday-emails", birthdayEmailsRouter()); return router; } diff --git a/src/routes/internalBirthday.ts b/src/routes/internalBirthday.ts deleted file mode 100644 index dba0387..0000000 --- a/src/routes/internalBirthday.ts +++ /dev/null @@ -1,27 +0,0 @@ -import { Router } from "express"; -import * as birthdayCtrl from "../controllers/birthdayEmail.controller"; -import { requireInternalAuth } from "../middleware/internalAuth"; - -export default function internalBirthdayRouter(): Router { - const router = Router(); - - // Protect all internal birthday tracking endpoints with service-to-service auth - router.use(requireInternalAuth); - - // Initialize jobs idempotently - router.post("/initialize", birthdayCtrl.initializeJobs); - - // Atomically claim a batch of eligible jobs - router.post("/claim", birthdayCtrl.claimJobs); - - // Statistics - router.get("/stats", birthdayCtrl.getStats); - - // Report job send result - router.post("/:id/sent", birthdayCtrl.markJobSent); - - // Report job failure result - router.post("/:id/failed", birthdayCtrl.markJobFailed); - - return router; -} diff --git a/src/services/birthdayEmail.service.ts b/src/services/birthdayEmail.service.ts index f61b135..f3d5bb9 100644 --- a/src/services/birthdayEmail.service.ts +++ b/src/services/birthdayEmail.service.ts @@ -1,4 +1,5 @@ import prisma from "../db/client"; +import type { Prisma } from "../generated/prisma/client"; import { ApiError } from "../utils/apiError"; import crypto from "crypto"; @@ -20,17 +21,30 @@ export interface JobStats { total: number; } -/** Retry delays in milliseconds based on attempt count */ -const RETRY_DELAYS_MS: Record = { - 1: 5 * 60 * 1000, // 5 min - 2: 15 * 60 * 1000, // 15 min - 3: 30 * 60 * 1000, // 30 min - 4: 120 * 60 * 1000, // 120 min -}; +export const ERROR_CLASSIFICATIONS = [ + "TRANSIENT", + "PERMANENT", + "CONFIG", + "QUOTA", +] as const; +export type ErrorClassification = (typeof ERROR_CLASSIFICATIONS)[number]; -const QUOTA_RETRY_DELAY_MS = 4 * 60 * 60 * 1000; // 4 hours for quota exhaustion +export interface FailedJobResult { + status: "FAILED" | "DEAD"; + attemptCount: number; + nextRetryAt: Date | null; +} + +const MINUTE_MS = 60 * 1000; + +/** Retry delay after the Nth failed attempt (index 0 = after attempt 1). */ +const RETRY_DELAYS_MS = [5, 15, 30, 120].map((m) => m * MINUTE_MS); +const QUOTA_RETRY_DELAY_MS = 4 * 60 * MINUTE_MS; const MAX_ATTEMPTS = 5; +export const DEFAULT_BATCH_SIZE = 10; +export const DEFAULT_LEASE_MS = 5 * MINUTE_MS; + /** * Sanitize error message to ensure no secrets, keys, or bearer tokens are stored. */ @@ -44,6 +58,11 @@ function sanitizeErrorMessage(msg: string): string { .substring(0, 500); } +function retryDelayFor(attemptCount: number): number { + const index = Math.min(Math.max(attemptCount, 1), RETRY_DELAYS_MS.length) - 1; + return RETRY_DELAYS_MS[index]; +} + /** * Initialize birthday email jobs for a given year and list of member IDs idempotently. */ @@ -139,10 +158,14 @@ export async function initializeBirthdayJobs( * - FAILED and nextRetryAt <= now * - PROCESSING and lease expired (lockedUntil < now) * Prevents race conditions with FOR UPDATE SKIP LOCKED. + * + * An expired PROCESSING lease means the previous worker never reported a + * result (for example, it crashed), so that lost attempt is counted. A job + * that runs out of attempts this way is moved to DEAD instead of reclaimed. */ export async function claimBirthdayJobs( - batchSize = 10, - leaseDurationMs = 300000 + batchSize = DEFAULT_BATCH_SIZE, + leaseDurationMs = DEFAULT_LEASE_MS ): Promise { const safeBatchSize = Math.max(1, Math.min(batchSize, 50)); const safeLeaseMs = Math.max(10000, Math.min(leaseDurationMs, 3600000)); @@ -150,8 +173,10 @@ export async function claimBirthdayJobs( return await prisma.$transaction(async (tx) => { // Select eligible jobs with row-level locks, skipping already-locked rows - const eligible = await tx.$queryRaw>` - SELECT id FROM "BirthdayEmailLog" + const eligible = await tx.$queryRaw< + Array<{ id: string; status: string; attemptCount: number }> + >` + SELECT id, "status", "attemptCount" FROM "BirthdayEmailLog" WHERE ( "status" = 'PENDING'::"BirthdayEmailStatus" OR ("status" = 'FAILED'::"BirthdayEmailStatus" AND ("nextRetryAt" IS NULL OR "nextRetryAt" <= ${now})) OR @@ -170,11 +195,32 @@ export async function claimBirthdayJobs( const claimed: ClaimedJob[] = []; for (const row of eligible) { + let attemptCount = row.attemptCount; + + if (row.status === "PROCESSING") { + attemptCount += 1; + if (attemptCount >= MAX_ATTEMPTS) { + await tx.birthdayEmailLog.update({ + where: { id: row.id }, + data: { + status: "DEAD", + attemptCount, + lastError: "Lease expired: worker never reported a result", + lockToken: null, + lockedUntil: null, + nextRetryAt: null, + }, + }); + continue; + } + } + const lockToken = crypto.randomUUID(); const updated = await tx.birthdayEmailLog.update({ where: { id: row.id }, data: { status: "PROCESSING", + attemptCount, lockToken, lockedUntil, }, @@ -183,78 +229,82 @@ export async function claimBirthdayJobs( memberId: true, birthdayYear: true, attemptCount: true, - lockToken: true, }, }); - if (updated.lockToken) { - claimed.push({ - id: updated.id, - memberId: updated.memberId, - birthdayYear: updated.birthdayYear, - attemptCount: updated.attemptCount, - lockToken: updated.lockToken, - }); - } + claimed.push({ ...updated, lockToken }); } return claimed; }); } +async function findJobOrThrow(id: string) { + const record = await prisma.birthdayEmailLog.findUnique({ where: { id } }); + if (!record) { + throw new ApiError("Birthday email record not found", 404); + } + return record; +} + +function leaseConflict(): ApiError { + return new ApiError( + "Invalid or expired lock token: job is no longer owned by caller", + 409 + ); +} + +/** + * Write a job result only if the caller still holds the lease. + * The lease check and the write are one statement, so a worker whose lease + * expired cannot overwrite a job that another worker has since claimed. + */ +async function updateIfLeaseOwned( + id: string, + lockToken: string, + data: Prisma.BirthdayEmailLogUpdateManyMutationInput +): Promise { + const { count } = await prisma.birthdayEmailLog.updateMany({ + where: { id, lockToken, status: "PROCESSING" }, + data, + }); + if (count === 0) { + throw leaseConflict(); + } +} + /** * Mark a claimed birthday email job as SENT. - * Validates the lockToken to guard against stale workers. + * Returns `alreadySent: true` when the job had already been marked as sent. */ export async function recordBirthdayJobSent( id: string, lockToken: string, providerMessageId?: string, - sentAt?: string -): Promise<{ success: boolean; message: string }> { - if (!id || typeof id !== "string") { - throw new ApiError("Missing tracking record id", 400); - } - if (!lockToken || typeof lockToken !== "string") { - throw new ApiError("Missing lockToken", 400); - } - - const record = await prisma.birthdayEmailLog.findUnique({ - where: { id }, - }); - - if (!record) { - throw new ApiError("Birthday email record not found", 404); - } + sentAt: Date = new Date() +): Promise<{ alreadySent: boolean }> { + const record = await findJobOrThrow(id); - // Idempotent duplicate check: already SENT with provider ID if (record.status === "SENT") { - return { success: true, message: "Record was already marked as sent" }; + return { alreadySent: true }; } - // Validate lease ownership if (record.status !== "PROCESSING" || record.lockToken !== lockToken) { - throw new ApiError( - "Invalid or expired lock token: job is no longer owned by caller", - 409 - ); + throw leaseConflict(); } - await prisma.birthdayEmailLog.update({ - where: { id }, - data: { - status: "SENT", - providerMessageId: providerMessageId ?? record.providerMessageId, - sentAt: sentAt ? new Date(sentAt) : new Date(), - lockToken: null, - lockedUntil: null, - nextRetryAt: null, - attemptCount: record.attemptCount + 1, - lastError: null, - }, + await updateIfLeaseOwned(id, lockToken, { + status: "SENT", + providerMessageId: providerMessageId ?? record.providerMessageId, + sentAt, + lockToken: null, + lockedUntil: null, + nextRetryAt: null, + attemptCount: record.attemptCount + 1, + lastError: null, }); - return { success: true, message: "Marked as sent" }; + return { alreadySent: false }; } /** @@ -264,78 +314,46 @@ export async function recordBirthdayJobSent( export async function recordBirthdayJobFailed( id: string, lockToken: string, - errorClassification: string, + errorClassification: ErrorClassification, errorMessage: string, isRetryable: boolean -): Promise<{ - success: boolean; - status: "FAILED" | "DEAD"; - attemptCount: number; - nextRetryAt: Date | null; -}> { - if (!id || typeof id !== "string") { - throw new ApiError("Missing tracking record id", 400); - } - if (!lockToken || typeof lockToken !== "string") { - throw new ApiError("Missing lockToken", 400); - } - - const record = await prisma.birthdayEmailLog.findUnique({ - where: { id }, - }); - - if (!record) { - throw new ApiError("Birthday email record not found", 404); - } +): Promise { + const record = await findJobOrThrow(id); if (record.status !== "PROCESSING" || record.lockToken !== lockToken) { - throw new ApiError( - "Invalid or expired lock token: job is no longer owned by caller", - 409 - ); + throw leaseConflict(); } - const newAttemptCount = record.attemptCount + 1; - const sanitized = sanitizeErrorMessage(errorMessage); - + const attemptCount = record.attemptCount + 1; const isPermanent = !isRetryable || errorClassification === "PERMANENT" || errorClassification === "CONFIG"; - let newStatus: "FAILED" | "DEAD"; + let status: "FAILED" | "DEAD"; let nextRetryAt: Date | null = null; - if (isPermanent || newAttemptCount >= MAX_ATTEMPTS) { - newStatus = "DEAD"; - nextRetryAt = null; + if (isPermanent || attemptCount >= MAX_ATTEMPTS) { + status = "DEAD"; } else { - newStatus = "FAILED"; - let delayMs = RETRY_DELAYS_MS[newAttemptCount] || 120 * 60 * 1000; - if (errorClassification === "QUOTA") { - delayMs = QUOTA_RETRY_DELAY_MS; - } + status = "FAILED"; + const delayMs = + errorClassification === "QUOTA" + ? QUOTA_RETRY_DELAY_MS + : retryDelayFor(attemptCount); nextRetryAt = new Date(Date.now() + delayMs); } - await prisma.birthdayEmailLog.update({ - where: { id }, - data: { - status: newStatus, - attemptCount: newAttemptCount, - nextRetryAt, - lastError: sanitized, - lockToken: null, - lockedUntil: null, - }, + await updateIfLeaseOwned(id, lockToken, { + status, + attemptCount, + nextRetryAt, + lastError: sanitizeErrorMessage(errorMessage), + lockToken: null, + lockedUntil: null, }); - return { - success: true, - status: newStatus, - attemptCount: newAttemptCount, - nextRetryAt, - }; + return { status, attemptCount, nextRetryAt }; } /** diff --git a/tests/BirthdayEmail.test.ts b/tests/BirthdayEmail.test.ts index 7e5626a..7106e2c 100644 --- a/tests/BirthdayEmail.test.ts +++ b/tests/BirthdayEmail.test.ts @@ -3,6 +3,7 @@ import * as birthdayService from "../src/services/birthdayEmail.service"; import * as birthdayCtrl from "../src/controllers/birthdayEmail.controller"; import { requireInternalAuth } from "../src/middleware/internalAuth"; import prisma from "../src/db/client"; +import config from "../src/config"; import { ApiError } from "../src/utils/apiError"; jest.mock("../src/db/client", () => { @@ -16,6 +17,7 @@ jest.mock("../src/db/client", () => { findUnique: jest.fn(), createMany: jest.fn(), update: jest.fn(), + updateMany: jest.fn(), groupBy: jest.fn(), count: jest.fn(), }, @@ -29,17 +31,38 @@ jest.mock("../src/db/client", () => { }; }); -describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { - const originalEnv = process.env; +const mockResponse = (): Response => { + const res = {} as Response; + res.status = jest.fn().mockReturnValue(res); + res.json = jest.fn().mockReturnValue(res); + return res; +}; + +/** Run claimBirthdayJobs against a fake transaction returning the given rows. */ +function mockClaimTransaction( + rows: Array<{ id: string; status: string; attemptCount: number }> +) { + const update = jest.fn().mockImplementation(({ where, data }) => + Promise.resolve({ + id: where.id, + memberId: `member-of-${where.id}`, + birthdayYear: 2026, + attemptCount: data.attemptCount, + }) + ); + (prisma.$transaction as jest.Mock).mockImplementationOnce(async (callback) => + callback({ + $queryRaw: jest.fn().mockResolvedValue(rows), + birthdayEmailLog: { update }, + }) + ); + return update; +} +describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { beforeEach(() => { jest.clearAllMocks(); - process.env = { ...originalEnv }; - process.env.INTERNAL_SERVICE_SECRET = "super-secret-service-token"; - }); - - afterAll(() => { - process.env = originalEnv; + config.INTERNAL_SERVICE_SECRET = "super-secret-service-token"; }); // ────────────────────────────────────────────── @@ -95,72 +118,76 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { // 3. Two concurrent workers cannot claim the same active job // ────────────────────────────────────────────── it("3. Atomic job claim locks jobs and prevents concurrent workers from claiming the same active job", async () => { - const jobRow = { id: "job-1" }; - - // Transaction executes queryRaw and update - (prisma.$transaction as jest.Mock).mockImplementation(async (callback) => { - const tx = { - $queryRaw: jest.fn().mockResolvedValue([jobRow]), - birthdayEmailLog: { - update: jest.fn().mockResolvedValue({ - id: "job-1", - memberId: "m-1", - birthdayYear: 2026, - attemptCount: 0, - lockToken: "token-worker-1", - }), - }, - }; - return await callback(tx); - }); + mockClaimTransaction([{ id: "job-1", status: "PENDING", attemptCount: 0 }]); const claimed = await birthdayService.claimBirthdayJobs(10, 300000); expect(claimed).toHaveLength(1); expect(claimed[0].id).toBe("job-1"); - expect(claimed[0].lockToken).toBeDefined(); + expect(claimed[0].attemptCount).toBe(0); + expect(claimed[0].lockToken).toEqual(expect.any(String)); // Second worker claims concurrently, queryRaw returns empty (already locked/skipped) - (prisma.$transaction as jest.Mock).mockImplementationOnce(async (callback) => { - const tx = { - $queryRaw: jest.fn().mockResolvedValue([]), - birthdayEmailLog: { update: jest.fn() }, - }; - return await callback(tx); - }); + mockClaimTransaction([]); const secondClaim = await birthdayService.claimBirthdayJobs(10, 300000); expect(secondClaim).toHaveLength(0); }); // ────────────────────────────────────────────── - // 4. Expired PROCESSING jobs can be recovered + // 4. Expired PROCESSING jobs can be recovered, and the lost attempt counts // ────────────────────────────────────────────── - it("4. Expired PROCESSING jobs can be recovered by claim query", async () => { - (prisma.$transaction as jest.Mock).mockImplementation(async (callback) => { - const tx = { - $queryRaw: jest.fn().mockImplementation((strings, ...values) => { - // Verify query includes lease expiration check: lockedUntil < now - const queryText = strings.join(""); - expect(queryText).toContain('"lockedUntil" <'); - return Promise.resolve([{ id: "expired-job-1" }]); + it("4. Expired PROCESSING jobs are reclaimed and the lost attempt is counted", async () => { + let queryText = ""; + const update = jest.fn().mockImplementation(({ where, data }) => + Promise.resolve({ + id: where.id, + memberId: "m-expired", + birthdayYear: 2026, + attemptCount: data.attemptCount, + }) + ); + (prisma.$transaction as jest.Mock).mockImplementationOnce(async (callback) => + callback({ + $queryRaw: jest.fn().mockImplementation((strings: string[]) => { + queryText = strings.join(""); + return Promise.resolve([ + { id: "expired-job-1", status: "PROCESSING", attemptCount: 1 }, + ]); }), - birthdayEmailLog: { - update: jest.fn().mockResolvedValue({ - id: "expired-job-1", - memberId: "m-expired", - birthdayYear: 2026, - attemptCount: 1, - lockToken: "new-recovery-token", - }), - }, - }; - return await callback(tx); - }); + birthdayEmailLog: { update }, + }) + ); const claimed = await birthdayService.claimBirthdayJobs(5, 300000); + expect(queryText).toContain('"lockedUntil" <'); expect(claimed).toHaveLength(1); expect(claimed[0].id).toBe("expired-job-1"); - expect(claimed[0].lockToken).toBe("new-recovery-token"); + expect(claimed[0].attemptCount).toBe(2); + expect(update).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ status: "PROCESSING", attemptCount: 2 }), + }) + ); + }); + + it("4b. An expired PROCESSING job on its last attempt goes DEAD instead of being reclaimed", async () => { + const update = mockClaimTransaction([ + { id: "crashy-job", status: "PROCESSING", attemptCount: 4 }, + { id: "fresh-job", status: "PENDING", attemptCount: 0 }, + ]); + + const claimed = await birthdayService.claimBirthdayJobs(10, 300000); + expect(claimed.map((j) => j.id)).toEqual(["fresh-job"]); + expect(update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: "crashy-job" }, + data: expect.objectContaining({ + status: "DEAD", + attemptCount: 5, + lockToken: null, + }), + }) + ); }); // ────────────────────────────────────────────── @@ -175,26 +202,10 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { attemptCount: 2, }); - // Stale worker tries to report sent with old token await expect( - birthdayService.recordBirthdayJobSent( - "job-conflict", - "stale-old-token", - "msg-123" - ) - ).rejects.toThrow(ApiError); - - try { - await birthdayService.recordBirthdayJobSent( - "job-conflict", - "stale-old-token", - "msg-123" - ); - } catch (err: any) { - expect(err.statusCode).toBe(409); - } + birthdayService.recordBirthdayJobSent("job-conflict", "stale-old-token", "msg-123") + ).rejects.toMatchObject({ statusCode: 409 }); - // Stale worker tries to report failed with old token await expect( birthdayService.recordBirthdayJobFailed( "job-conflict", @@ -203,7 +214,35 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { "network timeout", true ) - ).rejects.toThrow(ApiError); + ).rejects.toMatchObject({ statusCode: 409 }); + + expect(prisma.birthdayEmailLog.updateMany).not.toHaveBeenCalled(); + }); + + it("5b. Losing the lease between the read and the write is rejected (409 conflict)", async () => { + // The read still shows our lease... + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-race", + status: "PROCESSING", + lockToken: "my-token", + attemptCount: 1, + }); + // ...but another worker reclaimed it before our write landed. + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + + await expect( + birthdayService.recordBirthdayJobSent("job-race", "my-token", "msg-1") + ).rejects.toMatchObject({ statusCode: 409 }); + + await expect( + birthdayService.recordBirthdayJobFailed("job-race", "my-token", "TRANSIENT", "boom", true) + ).rejects.toMatchObject({ statusCode: 409 }); + + expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: "job-race", lockToken: "my-token", status: "PROCESSING" }, + }) + ); }); // ────────────────────────────────────────────── @@ -217,35 +256,41 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { attemptCount: 0, providerMessageId: null, }); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); - (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({ - id: "job-success", - status: "SENT", - providerMessageId: "resend-msg-999", - sentAt: new Date("2026-10-21T09:05:00.000Z"), - }); - + const sentAt = new Date("2026-10-21T09:05:00.000Z"); const res = await birthdayService.recordBirthdayJobSent( "job-success", "valid-token-abc", "resend-msg-999", - "2026-10-21T09:05:00.000Z" + sentAt ); - expect(res.success).toBe(true); - expect(prisma.birthdayEmailLog.update).toHaveBeenCalledWith( - expect.objectContaining({ - where: { id: "job-success" }, - data: expect.objectContaining({ - status: "SENT", - providerMessageId: "resend-msg-999", - sentAt: new Date("2026-10-21T09:05:00.000Z"), - lockToken: null, - lockedUntil: null, - attemptCount: 1, - }), - }) - ); + expect(res.alreadySent).toBe(false); + expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith({ + where: { id: "job-success", lockToken: "valid-token-abc", status: "PROCESSING" }, + data: expect.objectContaining({ + status: "SENT", + providerMessageId: "resend-msg-999", + sentAt, + lockToken: null, + lockedUntil: null, + attemptCount: 1, + }), + }); + }); + + it("6b. Reporting an already SENT job is a no-op", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-done", + status: "SENT", + lockToken: null, + attemptCount: 1, + }); + + const res = await birthdayService.recordBirthdayJobSent("job-done", "any-token"); + expect(res.alreadySent).toBe(true); + expect(prisma.birthdayEmailLog.updateMany).not.toHaveBeenCalled(); }); // ────────────────────────────────────────────── @@ -258,8 +303,7 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { lockToken: "tok-retry", attemptCount: 1, // attempt 1 failed, next will be attempt 2 }); - - (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({}); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); const res = await birthdayService.recordBirthdayJobFailed( "job-retryable", @@ -269,7 +313,6 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { true ); - expect(res.success).toBe(true); expect(res.status).toBe("FAILED"); expect(res.attemptCount).toBe(2); expect(res.nextRetryAt).toBeInstanceOf(Date); @@ -289,8 +332,7 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { lockToken: "tok-perm", attemptCount: 0, }); - - (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({}); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); const res = await birthdayService.recordBirthdayJobFailed( "job-perm", @@ -300,11 +342,10 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { false ); - expect(res.success).toBe(true); expect(res.status).toBe("DEAD"); expect(res.nextRetryAt).toBeNull(); expect(res.attemptCount).toBe(1); - expect(prisma.birthdayEmailLog.update).toHaveBeenCalledWith( + expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith( expect.objectContaining({ data: expect.objectContaining({ status: "DEAD", @@ -324,8 +365,7 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { lockToken: "tok-max", attemptCount: 4, // 5th attempt failing }); - - (prisma.birthdayEmailLog.update as jest.Mock).mockResolvedValue({}); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); const res = await birthdayService.recordBirthdayJobFailed( "job-max-attempt", @@ -335,7 +375,6 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { true ); - expect(res.success).toBe(true); expect(res.status).toBe("DEAD"); expect(res.attemptCount).toBe(5); expect(res.nextRetryAt).toBeNull(); @@ -346,45 +385,128 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { // ────────────────────────────────────────────── it("10. Unauthorized internal API requests are rejected by middleware", () => { const next: NextFunction = jest.fn(); - - // 10a: Missing secret - const reqMissing = { - headers: {}, - } as unknown as Request; const res = {} as Response; - expect(() => requireInternalAuth(reqMissing, res, next)).toThrow(ApiError); - try { - requireInternalAuth(reqMissing, res, next); - } catch (err: any) { - expect(err.statusCode).toBe(401); - } + // 10a: Missing secret + const reqMissing = { headers: {} } as unknown as Request; + expect(() => requireInternalAuth(reqMissing, res, next)).toThrow( + expect.objectContaining({ statusCode: 401 }) + ); - // 10b: Invalid secret + // 10b: Invalid secret (different length from the real one) const reqInvalid = { - headers: { - "x-service-secret": "wrong-secret-token", - }, + headers: { "x-service-secret": "wrong" }, } as unknown as Request; - - expect(() => requireInternalAuth(reqInvalid, res, next)).toThrow(ApiError); - try { - requireInternalAuth(reqInvalid, res, next); - } catch (err: any) { - expect(err.statusCode).toBe(401); - } + expect(() => requireInternalAuth(reqInvalid, res, next)).toThrow( + expect.objectContaining({ statusCode: 401 }) + ); + expect(next).not.toHaveBeenCalled(); // 10c: Valid secret const reqValid = { - headers: { - "x-service-secret": "super-secret-service-token", - }, + headers: { "x-service-secret": "super-secret-service-token" }, } as unknown as Request; - expect(() => requireInternalAuth(reqValid, res, next)).not.toThrow(); expect(next).toHaveBeenCalled(); }); + it("10d. Missing server secret is a 500", () => { + config.INTERNAL_SERVICE_SECRET = undefined; + const req = { headers: { "x-internal-secret": "anything" } } as unknown as Request; + expect(() => requireInternalAuth(req, {} as Response, jest.fn())).toThrow( + expect.objectContaining({ statusCode: 500 }) + ); + }); + + // ────────────────────────────────────────────── + // 11. Controller input validation and response shape + // ────────────────────────────────────────────── + describe("controller", () => { + it("rejects an invalid sentAt with 400", async () => { + const req = { + params: { id: "job-1" }, + body: { lockToken: "tok", sentAt: "not-a-date" }, + } as unknown as Request; + + await expect(birthdayCtrl.markJobSent(req, mockResponse())).rejects.toMatchObject({ + statusCode: 400, + }); + expect(prisma.birthdayEmailLog.findUnique).not.toHaveBeenCalled(); + }); + + it("rejects an unknown errorClassification with 400", async () => { + const req = { + params: { id: "job-1" }, + body: { lockToken: "tok", errorClassification: "permanent" }, + } as unknown as Request; + + await expect(birthdayCtrl.markJobFailed(req, mockResponse())).rejects.toBeInstanceOf( + ApiError + ); + expect(prisma.birthdayEmailLog.findUnique).not.toHaveBeenCalled(); + }); + + it("rejects a missing lockToken with 400", async () => { + const req = { params: { id: "job-1" }, body: {} } as unknown as Request; + + await expect(birthdayCtrl.markJobSent(req, mockResponse())).rejects.toMatchObject({ + statusCode: 400, + }); + }); + + it("wraps a failure result as { success: true, ... }", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-1", + status: "PROCESSING", + lockToken: "tok", + attemptCount: 0, + }); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + + const req = { + params: { id: "job-1" }, + body: { lockToken: "tok", errorClassification: "QUOTA", errorMessage: "429" }, + } as unknown as Request; + const res = mockResponse(); + + await birthdayCtrl.markJobFailed(req, res); + + expect(res.status).toHaveBeenCalledWith(200); + expect(res.json).toHaveBeenCalledWith({ + success: true, + status: "FAILED", + attemptCount: 1, + nextRetryAt: expect.any(Date), + }); + }); + + it("passes a valid sentAt through as a Date", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-1", + status: "PROCESSING", + lockToken: "tok", + attemptCount: 0, + providerMessageId: null, + }); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + + const req = { + params: { id: "job-1" }, + body: { lockToken: "tok", sentAt: "2026-10-21T09:05:00.000Z" }, + } as unknown as Request; + const res = mockResponse(); + + await birthdayCtrl.markJobSent(req, res); + + expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ sentAt: new Date("2026-10-21T09:05:00.000Z") }), + }) + ); + expect(res.json).toHaveBeenCalledWith({ success: true, message: "Marked as sent" }); + }); + }); + // ────────────────────────────────────────────── // Stats endpoint // ────────────────────────────────────────────── From 0f5bfdf9d045351003aa228a601cfff98e920339 Mon Sep 17 00:00:00 2001 From: Sarvesh Shahane Date: Wed, 21 Oct 2026 21:04:33 +0530 Subject: [PATCH 3/4] fix: accept UNCERTAIN failures and tighten birthday API input - Accept the UNCERTAIN error classification sent by the COC-Members birthday worker (provider timeout after dispatch); retry it like TRANSIENT. The previous commit rejected it with a 400. - Validate providerMessageId as string, null or missing (400 otherwise) - Accept only the x-internal-secret header; the worker already sends it - Match sibling controllers: export const handlers, single service import --- src/controllers/birthdayEmail.controller.ts | 39 ++++++---- src/middleware/internalAuth.ts | 13 +--- src/routes/birthdayEmails.ts | 8 +- src/services/birthdayEmail.service.ts | 6 ++ tests/BirthdayEmail.test.ts | 85 ++++++++++++++++++++- 5 files changed, 120 insertions(+), 31 deletions(-) diff --git a/src/controllers/birthdayEmail.controller.ts b/src/controllers/birthdayEmail.controller.ts index fdbf7b4..7232bda 100644 --- a/src/controllers/birthdayEmail.controller.ts +++ b/src/controllers/birthdayEmail.controller.ts @@ -1,11 +1,10 @@ import { Request, Response } from "express"; import * as birthdayEmailService from "../services/birthdayEmail.service"; -import { - ERROR_CLASSIFICATIONS, - ErrorClassification, -} from "../services/birthdayEmail.service"; import { ApiError } from "../utils/apiError"; +const { ERROR_CLASSIFICATIONS } = birthdayEmailService; +type ErrorClassification = birthdayEmailService.ErrorClassification; + function requireLockToken(body: Record): string { const { lockToken } = body; if (!lockToken || typeof lockToken !== "string") { @@ -23,6 +22,14 @@ function parseSentAt(value: unknown): Date | undefined { return date; } +function parseProviderMessageId(value: unknown): string | undefined { + if (value === undefined || value === null) return undefined; + if (typeof value !== "string") { + throw new ApiError("providerMessageId must be a string", 400); + } + return value; +} + function parseErrorClassification(value: unknown): ErrorClassification { if (value === undefined || value === null) return "TRANSIENT"; if (!ERROR_CLASSIFICATIONS.includes(value as ErrorClassification)) { @@ -37,7 +44,7 @@ function parseErrorClassification(value: unknown): ErrorClassification { /** * POST /api/v1/birthday-emails/initialize */ -export async function initializeJobs(req: Request, res: Response): Promise { +export const initializeJobs = async (req: Request, res: Response) => { const { birthdayYear, memberIds } = req.body || {}; if (typeof birthdayYear !== "number") { @@ -58,12 +65,12 @@ export async function initializeJobs(req: Request, res: Response): Promise message: "Birthday jobs initialized", data: result, }); -} +}; /** * POST /api/v1/birthday-emails/claim */ -export async function claimJobs(req: Request, res: Response): Promise { +export const claimJobs = async (req: Request, res: Response) => { const { batchSize, leaseDurationMs } = req.body || {}; const jobs = await birthdayEmailService.claimBirthdayJobs( @@ -76,20 +83,21 @@ export async function claimJobs(req: Request, res: Response): Promise { count: jobs.length, jobs, }); -} +}; /** * POST /api/v1/birthday-emails/:id/sent */ -export async function markJobSent(req: Request, res: Response): Promise { +export const markJobSent = async (req: Request, res: Response) => { const body = req.body || {}; const lockToken = requireLockToken(body); + const providerMessageId = parseProviderMessageId(body.providerMessageId); const sentAt = parseSentAt(body.sentAt); const { alreadySent } = await birthdayEmailService.recordBirthdayJobSent( req.params.id, lockToken, - body.providerMessageId, + providerMessageId, sentAt ); @@ -97,12 +105,12 @@ export async function markJobSent(req: Request, res: Response): Promise { success: true, message: alreadySent ? "Record was already marked as sent" : "Marked as sent", }); -} +}; /** * POST /api/v1/birthday-emails/:id/failed */ -export async function markJobFailed(req: Request, res: Response): Promise { +export const markJobFailed = async (req: Request, res: Response) => { const body = req.body || {}; const lockToken = requireLockToken(body); const errorClassification = parseErrorClassification(body.errorClassification); @@ -115,19 +123,20 @@ export async function markJobFailed(req: Request, res: Response): Promise typeof body.isRetryable === "boolean" ? body.isRetryable : true ); + // Flat shape is the contract the COC-Members birthday worker is typed against. res.status(200).json({ success: true, ...result, }); -} +}; /** * GET /api/v1/birthday-emails/stats */ -export async function getStats(_req: Request, res: Response): Promise { +export const getStats = async (_req: Request, res: Response) => { const stats = await birthdayEmailService.getBirthdayEmailStats(); res.status(200).json({ success: true, stats, }); -} +}; diff --git a/src/middleware/internalAuth.ts b/src/middleware/internalAuth.ts index ce3e282..c9a3b67 100644 --- a/src/middleware/internalAuth.ts +++ b/src/middleware/internalAuth.ts @@ -9,7 +9,7 @@ function sha256(value: string): Buffer { /** * Service-to-service authentication middleware for internal APIs. - * Compares the incoming secret header against INTERNAL_SERVICE_SECRET in constant time. + * Compares the `x-internal-secret` header against INTERNAL_SERVICE_SECRET in constant time. */ export function requireInternalAuth( req: Request, @@ -25,16 +25,9 @@ export function requireInternalAuth( ); } - const authHeader = req.headers["authorization"]; - let providedSecret = - (req.headers["x-internal-secret"] as string) || - (req.headers["x-service-secret"] as string); + const providedSecret = req.headers["x-internal-secret"]; - if (!providedSecret && authHeader && authHeader.startsWith("Bearer ")) { - providedSecret = authHeader.substring(7).trim(); - } - - if (!providedSecret) { + if (!providedSecret || typeof providedSecret !== "string") { throw new ApiError( "Unauthorized: Missing internal service secret header", 401 diff --git a/src/routes/birthdayEmails.ts b/src/routes/birthdayEmails.ts index cf8d13a..d5887c1 100644 --- a/src/routes/birthdayEmails.ts +++ b/src/routes/birthdayEmails.ts @@ -4,8 +4,7 @@ import { requireInternalAuth } from "../middleware/internalAuth"; /** * Internal, service-to-service API used by the birthday email worker. - * Every route requires the INTERNAL_SERVICE_SECRET, sent as an - * `x-internal-secret` / `x-service-secret` header or as a Bearer token. + * Every route requires the INTERNAL_SERVICE_SECRET in the `x-internal-secret` header. */ export default function birthdayEmailsRouter(): Router { const router = Router(); @@ -80,7 +79,7 @@ export default function birthdayEmailsRouter(): Router { * @apiParam (URL Params) {String} id Job ID. * * @apiBody {String} lockToken Token returned by /claim. (Required) - * @apiBody {String} [providerMessageId] Email provider's message ID. + * @apiBody {String} [providerMessageId] Email provider's message ID (null allowed). * @apiBody {String} [sentAt] ISO timestamp of the send. Defaults to now. * * @apiSuccess {Boolean} success Always true. @@ -104,7 +103,8 @@ export default function birthdayEmailsRouter(): Router { * @apiParam (URL Params) {String} id Job ID. * * @apiBody {String} lockToken Token returned by /claim. (Required) - * @apiBody {String="TRANSIENT","PERMANENT","CONFIG","QUOTA"} [errorClassification=TRANSIENT] Failure type. + * @apiBody {String="TRANSIENT","UNCERTAIN","PERMANENT","CONFIG","QUOTA"} [errorClassification=TRANSIENT] Failure type. + * UNCERTAIN (provider timed out after the request was sent) is retried like TRANSIENT. * @apiBody {String} [errorMessage] Error details. Secrets are redacted before storing. * @apiBody {Boolean} [isRetryable=true] Whether the failure can be retried. * diff --git a/src/services/birthdayEmail.service.ts b/src/services/birthdayEmail.service.ts index f3d5bb9..e4f2c63 100644 --- a/src/services/birthdayEmail.service.ts +++ b/src/services/birthdayEmail.service.ts @@ -21,8 +21,14 @@ export interface JobStats { total: number; } +/** + * UNCERTAIN means the request to the email provider timed out after it was + * sent, so the email may or may not have gone out. It is retried like + * TRANSIENT (the provider's idempotency key stops a duplicate send). + */ export const ERROR_CLASSIFICATIONS = [ "TRANSIENT", + "UNCERTAIN", "PERMANENT", "CONFIG", "QUOTA", diff --git a/tests/BirthdayEmail.test.ts b/tests/BirthdayEmail.test.ts index 7106e2c..6360518 100644 --- a/tests/BirthdayEmail.test.ts +++ b/tests/BirthdayEmail.test.ts @@ -395,7 +395,7 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { // 10b: Invalid secret (different length from the real one) const reqInvalid = { - headers: { "x-service-secret": "wrong" }, + headers: { "x-internal-secret": "wrong" }, } as unknown as Request; expect(() => requireInternalAuth(reqInvalid, res, next)).toThrow( expect.objectContaining({ statusCode: 401 }) @@ -404,12 +404,27 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { // 10c: Valid secret const reqValid = { - headers: { "x-service-secret": "super-secret-service-token" }, + headers: { "x-internal-secret": "super-secret-service-token" }, } as unknown as Request; expect(() => requireInternalAuth(reqValid, res, next)).not.toThrow(); expect(next).toHaveBeenCalled(); }); + it("10e. Only the x-internal-secret header is accepted", () => { + const next: NextFunction = jest.fn(); + const secret = "super-secret-service-token"; + for (const headers of [ + { "x-service-secret": secret }, + { authorization: `Bearer ${secret}` }, + ]) { + const req = { headers } as unknown as Request; + expect(() => requireInternalAuth(req, {} as Response, next)).toThrow( + expect.objectContaining({ statusCode: 401 }) + ); + } + expect(next).not.toHaveBeenCalled(); + }); + it("10d. Missing server secret is a 500", () => { config.INTERNAL_SERVICE_SECRET = undefined; const req = { headers: { "x-internal-secret": "anything" } } as unknown as Request; @@ -446,6 +461,72 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { expect(prisma.birthdayEmailLog.findUnique).not.toHaveBeenCalled(); }); + it("accepts UNCERTAIN and retries it like TRANSIENT", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-1", + status: "PROCESSING", + lockToken: "tok", + attemptCount: 0, + }); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + + const req = { + params: { id: "job-1" }, + body: { + lockToken: "tok", + errorClassification: "UNCERTAIN", + errorMessage: "Network timeout: ETIMEDOUT", + isRetryable: true, + }, + } as unknown as Request; + const res = mockResponse(); + + await birthdayCtrl.markJobFailed(req, res); + + const body = (res.json as jest.Mock).mock.calls[0][0]; + expect(body.status).toBe("FAILED"); + // Attempt 1 backoff is 5 minutes + const diffMs = body.nextRetryAt.getTime() - Date.now(); + expect(diffMs).toBeGreaterThan(4 * 60 * 1000); + expect(diffMs).toBeLessThanOrEqual(5 * 60 * 1000); + }); + + it("accepts a null providerMessageId and keeps the stored one", async () => { + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ + id: "job-1", + status: "PROCESSING", + lockToken: "tok", + attemptCount: 0, + providerMessageId: null, + }); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + + const req = { + params: { id: "job-1" }, + body: { lockToken: "tok", providerMessageId: null }, + } as unknown as Request; + + await birthdayCtrl.markJobSent(req, mockResponse()); + + expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ status: "SENT", providerMessageId: null }), + }) + ); + }); + + it("rejects a non-string providerMessageId with 400", async () => { + const req = { + params: { id: "job-1" }, + body: { lockToken: "tok", providerMessageId: 12345 }, + } as unknown as Request; + + await expect(birthdayCtrl.markJobSent(req, mockResponse())).rejects.toMatchObject({ + statusCode: 400, + }); + expect(prisma.birthdayEmailLog.findUnique).not.toHaveBeenCalled(); + }); + it("rejects a missing lockToken with 400", async () => { const req = { params: { id: "job-1" }, body: {} } as unknown as Request; From f1055fd21374bdd2c6c67aeb39a91b4ea8b33ed0 Mon Sep 17 00:00:00 2001 From: Sarvesh Shahane Date: Wed, 21 Oct 2026 21:11:51 +0530 Subject: [PATCH 4/4] refactor: simplify birthday email service - initialize: rely on the unique (memberId, birthdayYear) index with createMany skipDuplicates instead of pre-loading existing jobs; move the email check into the member query - /sent: write first with the lease-guarded updateMany and only read on a miss (404 / already sent / 409). Saves a query per send and makes overlapping /sent calls return alreadySent instead of 409 - stats: replace the status switch with a lookup map - drop controller checks the service already does - derive ClaimedJob from the Prisma model --- src/controllers/birthdayEmail.controller.ts | 9 +- src/services/birthdayEmail.service.ts | 136 +++++++------------ tests/BirthdayEmail.test.ts | 138 +++++++++++--------- 3 files changed, 124 insertions(+), 159 deletions(-) diff --git a/src/controllers/birthdayEmail.controller.ts b/src/controllers/birthdayEmail.controller.ts index 7232bda..bd5df91 100644 --- a/src/controllers/birthdayEmail.controller.ts +++ b/src/controllers/birthdayEmail.controller.ts @@ -45,16 +45,9 @@ function parseErrorClassification(value: unknown): ErrorClassification { * POST /api/v1/birthday-emails/initialize */ export const initializeJobs = async (req: Request, res: Response) => { + // birthdayYear and memberIds are validated by the service const { birthdayYear, memberIds } = req.body || {}; - if (typeof birthdayYear !== "number") { - throw new ApiError("birthdayYear is required and must be a number", 400); - } - - if (!Array.isArray(memberIds)) { - throw new ApiError("memberIds is required and must be an array of strings", 400); - } - const result = await birthdayEmailService.initializeBirthdayJobs( birthdayYear, memberIds diff --git a/src/services/birthdayEmail.service.ts b/src/services/birthdayEmail.service.ts index e4f2c63..e46bbb9 100644 --- a/src/services/birthdayEmail.service.ts +++ b/src/services/birthdayEmail.service.ts @@ -1,15 +1,12 @@ import prisma from "../db/client"; -import type { Prisma } from "../generated/prisma/client"; +import type { BirthdayEmailLog, Prisma } from "../generated/prisma/client"; import { ApiError } from "../utils/apiError"; import crypto from "crypto"; -export interface ClaimedJob { - id: string; - memberId: string; - birthdayYear: number; - attemptCount: number; - lockToken: string; -} +export type ClaimedJob = Pick< + BirthdayEmailLog, + "id" | "memberId" | "birthdayYear" | "attemptCount" +> & { lockToken: string }; export interface JobStats { pending: number; @@ -96,64 +93,33 @@ export async function initializeBirthdayJobs( throw new ApiError("No valid member IDs provided", 400); } - // Fetch only eligible members: approved, not ghosted, has birth_date, has valid email + // Eligible: approved, not ghosted, has a birth date and an email address const eligibleMembers = await prisma.member.findMany({ where: { id: { in: cleanIds }, isApproved: true, isGhosted: false, birth_date: { not: null }, + email: { contains: "@" }, }, - select: { - id: true, - email: true, - }, + select: { id: true }, }); - const validEligible = eligibleMembers.filter( - (m) => typeof m.email === "string" && m.email.includes("@") - ); - - if (validEligible.length === 0) { + if (eligibleMembers.length === 0) { return { created: 0, existing: 0, totalEligible: 0 }; } - const eligibleIds = validEligible.map((m) => m.id); - - // Check existing logs to preserve state and avoid resetting SENT or attempt counts - const existingLogs = await prisma.birthdayEmailLog.findMany({ - where: { - birthdayYear, - memberId: { in: eligibleIds }, - }, - select: { - memberId: true, - }, + // The unique (memberId, birthdayYear) index makes skipDuplicates leave + // existing jobs untouched, so SENT jobs and attempt counts are never reset. + const { count: created } = await prisma.birthdayEmailLog.createMany({ + data: eligibleMembers.map((m) => ({ memberId: m.id, birthdayYear })), + skipDuplicates: true, }); - const existingMemberIdSet = new Set(existingLogs.map((l) => l.memberId)); - const newMemberIds = eligibleIds.filter((id) => !existingMemberIdSet.has(id)); - - let createdCount = 0; - if (newMemberIds.length > 0) { - const recordsToCreate = newMemberIds.map((mId) => ({ - memberId: mId, - birthdayYear, - status: "PENDING" as const, - attemptCount: 0, - })); - - const result = await prisma.birthdayEmailLog.createMany({ - data: recordsToCreate, - skipDuplicates: true, - }); - createdCount = result.count; - } - return { - created: createdCount, - existing: existingLogs.length, - totalEligible: validEligible.length, + created, + existing: eligibleMembers.length - created, + totalEligible: eligibleMembers.length, }; } @@ -264,19 +230,18 @@ function leaseConflict(): ApiError { * Write a job result only if the caller still holds the lease. * The lease check and the write are one statement, so a worker whose lease * expired cannot overwrite a job that another worker has since claimed. + * Returns false when the caller no longer holds the lease. */ async function updateIfLeaseOwned( id: string, lockToken: string, data: Prisma.BirthdayEmailLogUpdateManyMutationInput -): Promise { +): Promise { const { count } = await prisma.birthdayEmailLog.updateMany({ where: { id, lockToken, status: "PROCESSING" }, data, }); - if (count === 0) { - throw leaseConflict(); - } + return count > 0; } /** @@ -289,28 +254,27 @@ export async function recordBirthdayJobSent( providerMessageId?: string, sentAt: Date = new Date() ): Promise<{ alreadySent: boolean }> { - const record = await findJobOrThrow(id); - - if (record.status === "SENT") { - return { alreadySent: true }; - } - - if (record.status !== "PROCESSING" || record.lockToken !== lockToken) { - throw leaseConflict(); - } - - await updateIfLeaseOwned(id, lockToken, { + const updated = await updateIfLeaseOwned(id, lockToken, { status: "SENT", - providerMessageId: providerMessageId ?? record.providerMessageId, + providerMessageId, // undefined leaves any stored ID unchanged sentAt, lockToken: null, lockedUntil: null, nextRetryAt: null, - attemptCount: record.attemptCount + 1, + attemptCount: { increment: 1 }, lastError: null, }); - return { alreadySent: false }; + if (updated) { + return { alreadySent: false }; + } + + // Nothing written: work out why. + const record = await findJobOrThrow(id); + if (record.status === "SENT") { + return { alreadySent: true }; + } + throw leaseConflict(); } /** @@ -350,7 +314,7 @@ export async function recordBirthdayJobFailed( nextRetryAt = new Date(Date.now() + delayMs); } - await updateIfLeaseOwned(id, lockToken, { + const updated = await updateIfLeaseOwned(id, lockToken, { status, attemptCount, nextRetryAt, @@ -358,10 +322,24 @@ export async function recordBirthdayJobFailed( lockToken: null, lockedUntil: null, }); + if (!updated) { + throw leaseConflict(); + } return { status, attemptCount, nextRetryAt }; } +const STATS_KEY_BY_STATUS: Record< + BirthdayEmailLog["status"], + "pending" | "processing" | "sent" | "failed" | "dead" +> = { + PENDING: "pending", + PROCESSING: "processing", + SENT: "sent", + FAILED: "failed", + DEAD: "dead", +}; + /** * Fetch tracking statistics without exposing member emails. */ @@ -383,24 +361,8 @@ export async function getBirthdayEmailStats(): Promise { for (const group of countsByStatus) { const count = group._count.id; + stats[STATS_KEY_BY_STATUS[group.status]] = count; stats.total += count; - switch (group.status) { - case "PENDING": - stats.pending = count; - break; - case "PROCESSING": - stats.processing = count; - break; - case "SENT": - stats.sent = count; - break; - case "FAILED": - stats.failed = count; - break; - case "DEAD": - stats.dead = count; - break; - } } const dueForRetry = await prisma.birthdayEmailLog.count({ diff --git a/tests/BirthdayEmail.test.ts b/tests/BirthdayEmail.test.ts index 6360518..8b0aa13 100644 --- a/tests/BirthdayEmail.test.ts +++ b/tests/BirthdayEmail.test.ts @@ -70,50 +70,67 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { // ────────────────────────────────────────────── it("1. Initializing the same member/year job twice creates one record", async () => { const memberId = "member-123"; - (prisma.member.findMany as jest.Mock).mockResolvedValue([ - { id: memberId, email: "alice@example.com" }, - ]); + (prisma.member.findMany as jest.Mock).mockResolvedValue([{ id: memberId }]); - // First initialization: no existing log - (prisma.birthdayEmailLog.findMany as jest.Mock).mockResolvedValueOnce([]); - (prisma.birthdayEmailLog.createMany as jest.Mock).mockResolvedValueOnce({ count: 1 }); + // First run inserts; second run hits the unique index and is skipped + (prisma.birthdayEmailLog.createMany as jest.Mock) + .mockResolvedValueOnce({ count: 1 }) + .mockResolvedValueOnce({ count: 0 }); const firstRun = await birthdayService.initializeBirthdayJobs(2026, [memberId]); - expect(firstRun.created).toBe(1); - expect(firstRun.existing).toBe(0); - - // Second initialization: existing log found - (prisma.birthdayEmailLog.findMany as jest.Mock).mockResolvedValueOnce([ - { memberId, status: "PENDING" }, - ]); + expect(firstRun).toEqual({ created: 1, existing: 0, totalEligible: 1 }); const secondRun = await birthdayService.initializeBirthdayJobs(2026, [memberId]); - expect(secondRun.created).toBe(0); - expect(secondRun.existing).toBe(1); - expect(prisma.birthdayEmailLog.createMany).toHaveBeenCalledTimes(1); + expect(secondRun).toEqual({ created: 0, existing: 1, totalEligible: 1 }); + + expect(prisma.birthdayEmailLog.createMany).toHaveBeenCalledWith({ + data: [{ memberId, birthdayYear: 2026 }], + skipDuplicates: true, + }); }); // ────────────────────────────────────────────── - // 2. Existing SENT records are not reset + // 2. Existing records are never reset, and only eligible members get jobs // ────────────────────────────────────────────── - it("2. Existing SENT records are not reset during initialization", async () => { - const memberId = "member-sent"; - (prisma.member.findMany as jest.Mock).mockResolvedValue([ - { id: memberId, email: "sent@example.com" }, - ]); + it("2. Existing records are not reset during initialization", async () => { + (prisma.member.findMany as jest.Mock).mockResolvedValue([{ id: "member-sent" }]); + (prisma.birthdayEmailLog.createMany as jest.Mock).mockResolvedValue({ count: 0 }); - // Existing record is SENT - (prisma.birthdayEmailLog.findMany as jest.Mock).mockResolvedValueOnce([ - { memberId, status: "SENT", attemptCount: 1 }, - ]); - - const res = await birthdayService.initializeBirthdayJobs(2026, [memberId]); + const res = await birthdayService.initializeBirthdayJobs(2026, ["member-sent"]); expect(res.created).toBe(0); expect(res.existing).toBe(1); + // Only inserts with skipDuplicates; never updates existing rows expect(prisma.birthdayEmailLog.update).not.toHaveBeenCalled(); + expect(prisma.birthdayEmailLog.updateMany).not.toHaveBeenCalled(); + }); + + it("2b. Eligibility filter is applied in the member query", async () => { + (prisma.member.findMany as jest.Mock).mockResolvedValue([]); + + const res = await birthdayService.initializeBirthdayJobs(2026, ["m1", "m1", " "]); + expect(res).toEqual({ created: 0, existing: 0, totalEligible: 0 }); + expect(prisma.member.findMany).toHaveBeenCalledWith({ + where: { + id: { in: ["m1"] }, + isApproved: true, + isGhosted: false, + birth_date: { not: null }, + email: { contains: "@" }, + }, + select: { id: true }, + }); expect(prisma.birthdayEmailLog.createMany).not.toHaveBeenCalled(); }); + it("2c. Invalid initialize input is rejected with 400", async () => { + await expect( + birthdayService.initializeBirthdayJobs("2026" as unknown as number, ["m1"]) + ).rejects.toMatchObject({ statusCode: 400 }); + await expect( + birthdayService.initializeBirthdayJobs(2026, undefined as unknown as string[]) + ).rejects.toMatchObject({ statusCode: 400 }); + }); + // ────────────────────────────────────────────── // 3. Two concurrent workers cannot claim the same active job // ────────────────────────────────────────────── @@ -194,18 +211,25 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { // 5. A stale lock token cannot overwrite a newer worker's result // ────────────────────────────────────────────── it("5. A stale lock token cannot overwrite a newer worker's result (409 conflict)", async () => { - // Record in DB currently has 'newer-lock-token' + // Record in DB currently has 'newer-lock-token', so the guarded write matches nothing (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ id: "job-conflict", status: "PROCESSING", lockToken: "newer-lock-token", attemptCount: 2, }); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); await expect( birthdayService.recordBirthdayJobSent("job-conflict", "stale-old-token", "msg-123") ).rejects.toMatchObject({ statusCode: 409 }); + expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: "job-conflict", lockToken: "stale-old-token", status: "PROCESSING" }, + }) + ); + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockClear(); await expect( birthdayService.recordBirthdayJobFailed( "job-conflict", @@ -215,11 +239,11 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { true ) ).rejects.toMatchObject({ statusCode: 409 }); - + // /failed rejects on its pre-read, before writing expect(prisma.birthdayEmailLog.updateMany).not.toHaveBeenCalled(); }); - it("5b. Losing the lease between the read and the write is rejected (409 conflict)", async () => { + it("5b. /failed losing the lease between the read and the write is rejected (409 conflict)", async () => { // The read still shows our lease... (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ id: "job-race", @@ -230,32 +254,24 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { // ...but another worker reclaimed it before our write landed. (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); - await expect( - birthdayService.recordBirthdayJobSent("job-race", "my-token", "msg-1") - ).rejects.toMatchObject({ statusCode: 409 }); - await expect( birthdayService.recordBirthdayJobFailed("job-race", "my-token", "TRANSIENT", "boom", true) ).rejects.toMatchObject({ statusCode: 409 }); + }); - expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith( - expect.objectContaining({ - where: { id: "job-race", lockToken: "my-token", status: "PROCESSING" }, - }) - ); + it("5c. /sent for an unknown job is a 404", async () => { + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue(null); + + await expect( + birthdayService.recordBirthdayJobSent("missing", "tok") + ).rejects.toMatchObject({ statusCode: 404 }); }); // ────────────────────────────────────────────── // 6. Successful sends persist provider IDs and timestamps // ────────────────────────────────────────────── - it("6. Successful sends persist provider IDs and timestamps", async () => { - (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ - id: "job-success", - status: "PROCESSING", - lockToken: "valid-token-abc", - attemptCount: 0, - providerMessageId: null, - }); + it("6. Successful sends persist provider IDs and timestamps in one write", async () => { (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); const sentAt = new Date("2026-10-21T09:05:00.000Z"); @@ -275,12 +291,15 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { sentAt, lockToken: null, lockedUntil: null, - attemptCount: 1, + attemptCount: { increment: 1 }, }), }); + expect(prisma.birthdayEmailLog.findUnique).not.toHaveBeenCalled(); }); - it("6b. Reporting an already SENT job is a no-op", async () => { + it("6b. Reporting an already SENT job (including overlapping calls) returns alreadySent", async () => { + // The guarded write matches nothing because the job is no longer PROCESSING + (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ id: "job-done", status: "SENT", @@ -290,7 +309,6 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { const res = await birthdayService.recordBirthdayJobSent("job-done", "any-token"); expect(res.alreadySent).toBe(true); - expect(prisma.birthdayEmailLog.updateMany).not.toHaveBeenCalled(); }); // ────────────────────────────────────────────── @@ -491,14 +509,7 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { expect(diffMs).toBeLessThanOrEqual(5 * 60 * 1000); }); - it("accepts a null providerMessageId and keeps the stored one", async () => { - (prisma.birthdayEmailLog.findUnique as jest.Mock).mockResolvedValue({ - id: "job-1", - status: "PROCESSING", - lockToken: "tok", - attemptCount: 0, - providerMessageId: null, - }); + it("accepts a null providerMessageId and leaves the stored one unchanged", async () => { (prisma.birthdayEmailLog.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); const req = { @@ -508,11 +519,10 @@ describe("Central Backend: Birthday Email Tracking & Internal APIs", () => { await birthdayCtrl.markJobSent(req, mockResponse()); - expect(prisma.birthdayEmailLog.updateMany).toHaveBeenCalledWith( - expect.objectContaining({ - data: expect.objectContaining({ status: "SENT", providerMessageId: null }), - }) - ); + const { data } = (prisma.birthdayEmailLog.updateMany as jest.Mock).mock.calls[0][0]; + expect(data.status).toBe("SENT"); + // undefined tells Prisma to leave the column as is + expect(data.providerMessageId).toBeUndefined(); }); it("rejects a non-string providerMessageId with 400", async () => {