From fa2dc67165d70c0483a3ecfdc0aef29e497585d5 Mon Sep 17 00:00:00 2001 From: AddarshKS Date: Fri, 17 Jul 2026 01:28:21 +0000 Subject: [PATCH] Email Automation V1.5 (Fixed the Automated Email Schedule-Sender and the Email Schedule Fairness Problem) --- .../migration.sql | 17 + .../tests/scheduler-state-seed.ts | 20 + apps/backend/prisma/schema.prisma | 17 + .../automations/schedule/route.test.ts | 19 + .../internal/automations/schedule/route.ts | 62 +- .../automations/scheduled-run/route.ts | 82 --- .../lib/automations/scheduler-state.test.ts | 154 +++++ .../src/lib/automations/scheduler-state.ts | 176 +++++ .../src/lib/automations/scheduler.test.ts | 578 ++++++++--------- apps/backend/src/lib/automations/scheduler.ts | 601 ++++++++++-------- apps/backend/src/lib/email-rendering.test.tsx | 44 +- packages/shared/src/config/schema.ts | 7 + packages/shared/src/helpers/emails.ts | 7 + .../src/helpers/usage-email-template.ts | 48 ++ 14 files changed, 1146 insertions(+), 686 deletions(-) create mode 100644 apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/migration.sql create mode 100644 apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/tests/scheduler-state-seed.ts create mode 100644 apps/backend/src/app/api/latest/internal/automations/schedule/route.test.ts delete mode 100644 apps/backend/src/app/api/latest/internal/automations/scheduled-run/route.ts create mode 100644 apps/backend/src/lib/automations/scheduler-state.test.ts create mode 100644 apps/backend/src/lib/automations/scheduler-state.ts create mode 100644 packages/shared/src/helpers/usage-email-template.ts diff --git a/apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/migration.sql b/apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/migration.sql new file mode 100644 index 000000000..afc3841e7 --- /dev/null +++ b/apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/migration.sql @@ -0,0 +1,17 @@ +CREATE TABLE "AutomationSchedulerState" ( + "key" TEXT NOT NULL, + "completedTenancyCursor" UUID, + "activeTenancyId" UUID, + "completedRuleCursor" TEXT, + "activeRuleId" TEXT, + "nextSubjectCursor" UUID, + "leaseOwner" UUID, + "leaseExpiresAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "AutomationSchedulerState_pkey" PRIMARY KEY ("key") +); + +INSERT INTO "AutomationSchedulerState" ("key", "updatedAt") +VALUES ('usage-email-v1', CURRENT_TIMESTAMP); diff --git a/apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/tests/scheduler-state-seed.ts b/apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/tests/scheduler-state-seed.ts new file mode 100644 index 000000000..26f86efa6 --- /dev/null +++ b/apps/backend/prisma/migrations/20260717000000_add_automation_scheduler_state/tests/scheduler-state-seed.ts @@ -0,0 +1,20 @@ +import type { Sql } from "postgres"; +import { expect } from "vitest"; + +export const postMigration = async (sql: Sql) => { + const stateRows = await sql` + SELECT "key", "completedTenancyCursor", "activeTenancyId", "completedRuleCursor", + "activeRuleId", "nextSubjectCursor", "leaseOwner", "leaseExpiresAt" + FROM "AutomationSchedulerState" + `; + expect(Array.from(stateRows)).toEqual([{ + key: "usage-email-v1", + completedTenancyCursor: null, + activeTenancyId: null, + completedRuleCursor: null, + activeRuleId: null, + nextSubjectCursor: null, + leaseOwner: null, + leaseExpiresAt: null, + }]); +}; diff --git a/apps/backend/prisma/schema.prisma b/apps/backend/prisma/schema.prisma index 08dfbccae..6465826c9 100644 --- a/apps/backend/prisma/schema.prisma +++ b/apps/backend/prisma/schema.prisma @@ -1187,6 +1187,23 @@ model AutomationRuleExecutionState { @@index([tenancyId, ruleId, lastTriggeredAt], name: "AutomationRuleExecutionState_tenancy_rule_triggered_idx") } +model AutomationSchedulerState { + key String @id + + completedTenancyCursor String? @db.Uuid + activeTenancyId String? @db.Uuid + completedRuleCursor String? + activeRuleId String? + // Continuation cursor passed to the next source-adapter page. + nextSubjectCursor String? @db.Uuid + + leaseOwner String? @db.Uuid + leaseExpiresAt DateTime? + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt +} + model Conversation { id String @default(uuid()) @db.Uuid tenancyId String @db.Uuid diff --git a/apps/backend/src/app/api/latest/internal/automations/schedule/route.test.ts b/apps/backend/src/app/api/latest/internal/automations/schedule/route.test.ts new file mode 100644 index 000000000..8b9724694 --- /dev/null +++ b/apps/backend/src/app/api/latest/internal/automations/schedule/route.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from "vitest"; +import { maxDuration, parseAutomationScheduleBound } from "./route"; + +describe("automation schedule route bounds", () => { + it("keeps the work budget below the route runtime limit", () => { + expect(maxDuration).toBe(60); + expect(parseAutomationScheduleBound("45000", "max_duration_ms", 45_000)).toBe(45_000); + }); + + it("allows omitted and lower positive integer bounds", () => { + expect(parseAutomationScheduleBound(undefined, "limit", 100)).toBeUndefined(); + expect(parseAutomationScheduleBound("25", "limit", 100)).toBe(25); + }); + + it.each(["0", "101", "1.5", "01", "not-a-number"])("rejects an invalid internal bound: %s", (value) => { + expect(() => parseAutomationScheduleBound(value, "limit", 100)) + .toThrowError("limit must be an integer between 1 and 100"); + }); +}); diff --git a/apps/backend/src/app/api/latest/internal/automations/schedule/route.ts b/apps/backend/src/app/api/latest/internal/automations/schedule/route.ts index 585e4d4d1..d583f8b09 100644 --- a/apps/backend/src/app/api/latest/internal/automations/schedule/route.ts +++ b/apps/backend/src/app/api/latest/internal/automations/schedule/route.ts @@ -1,8 +1,9 @@ import { - discoverEnabledScheduledAutomationRules, - enqueueScheduledAutomationRuns, - normalizeScheduledAutomationDiscoveryLimit, - normalizeScheduledAutomationRunPageLimit, + runScheduledAutomations, + scheduledAutomationDiscoveryLimit, + scheduledAutomationMaxPages, + scheduledAutomationRunPageLimit, + scheduledAutomationWorkBudgetMs, } from "@/lib/automations/scheduler"; import { createSmartRouteHandler } from "@/route-handlers/smart-route-handler"; import { yupBoolean, yupNumber, yupObject, yupString, yupTuple } from "@hexclave/shared/dist/schema-fields"; @@ -11,11 +12,12 @@ import { StatusError } from "@hexclave/shared/dist/utils/errors"; export const dynamic = "force-dynamic"; export const fetchCache = "force-no-store"; +export const maxDuration = 60; export const GET = createSmartRouteHandler({ metadata: { - summary: "Schedule automation rule runs", - description: "Internal endpoint invoked by cron to enqueue bounded scheduled automation rule runs.", + summary: "Run scheduled automation rules", + description: "Internal endpoint invoked by cron to execute bounded scheduled automation work.", tags: ["Automations"], hidden: true, }, @@ -26,9 +28,10 @@ export const GET = createSmartRouteHandler({ authorization: yupTuple([yupString().defined()]).optional(), }).defined(), query: yupObject({ - cursor: yupString().optional(), max_tenancies: yupString().optional(), limit: yupString().optional(), + max_pages: yupString().optional(), + max_duration_ms: yupString().optional(), }).defined(), }), response: yupObject({ @@ -36,10 +39,14 @@ export const GET = createSmartRouteHandler({ bodyType: yupString().oneOf(["json"]).defined(), body: yupObject({ ok: yupBoolean().defined(), + status: yupString().oneOf(["ran", "lease-held"]).defined(), tenancies_scanned: yupNumber().integer().defined(), - rules_found: yupNumber().integer().defined(), - runs_enqueued: yupNumber().integer().defined(), - next_cursor: yupString().nullable().defined(), + rules_processed: yupNumber().integer().defined(), + pages_processed: yupNumber().integer().defined(), + evaluated_count: yupNumber().integer().defined(), + sent_count: yupNumber().integer().defined(), + suppressed_count: yupNumber().integer().defined(), + cycle_completed: yupBoolean().defined(), }).defined(), }), handler: async ({ auth, headers, query }) => { @@ -49,14 +56,11 @@ export const GET = createSmartRouteHandler({ throw new StatusError(401, "Unauthorized"); } - const discovery = await discoverEnabledScheduledAutomationRules({ - limit: parseOptionalPositiveInteger(query.max_tenancies, "max_tenancies"), - cursor: query.cursor, - }); - const enqueueResult = await enqueueScheduledAutomationRuns({ - targets: discovery.targets, - limit: parseOptionalPositiveInteger(query.limit, "limit"), - scheduledAt: new Date(), + const result = await runScheduledAutomations({ + maxTenancies: parseAutomationScheduleBound(query.max_tenancies, "max_tenancies", scheduledAutomationDiscoveryLimit), + pageLimit: parseAutomationScheduleBound(query.limit, "limit", scheduledAutomationRunPageLimit), + maxPages: parseAutomationScheduleBound(query.max_pages, "max_pages", scheduledAutomationMaxPages), + workBudgetMs: parseAutomationScheduleBound(query.max_duration_ms, "max_duration_ms", scheduledAutomationWorkBudgetMs), }); return { @@ -64,24 +68,26 @@ export const GET = createSmartRouteHandler({ bodyType: "json", body: { ok: true, - tenancies_scanned: discovery.scannedTenancyCount, - rules_found: discovery.targets.length, - runs_enqueued: enqueueResult.enqueuedCount, - next_cursor: discovery.nextCursor, + status: result.status, + tenancies_scanned: result.tenanciesScanned, + rules_processed: result.rulesProcessed, + pages_processed: result.pagesProcessed, + evaluated_count: result.evaluatedCount, + sent_count: result.sentCount, + suppressed_count: result.suppressedCount, + cycle_completed: result.cycleCompleted, }, }; }, }); -function parseOptionalPositiveInteger(value: string | null | undefined, label: "max_tenancies" | "limit") { +export function parseAutomationScheduleBound(value: string | null | undefined, label: string, maximum: number) { if (value == null || value === "") { return undefined; } const parsed = Number.parseInt(value, 10); - if (!Number.isFinite(parsed) || parsed < 1 || String(parsed) !== value) { - throw new StatusError(400, `${label} must be a positive integer`); + if (!Number.isFinite(parsed) || parsed < 1 || parsed > maximum || String(parsed) !== value) { + throw new StatusError(400, `${label} must be an integer between 1 and ${maximum}`); } - return label === "max_tenancies" - ? normalizeScheduledAutomationDiscoveryLimit(parsed) - : normalizeScheduledAutomationRunPageLimit(parsed); + return parsed; } diff --git a/apps/backend/src/app/api/latest/internal/automations/scheduled-run/route.ts b/apps/backend/src/app/api/latest/internal/automations/scheduled-run/route.ts deleted file mode 100644 index b622c811b..000000000 --- a/apps/backend/src/app/api/latest/internal/automations/scheduled-run/route.ts +++ /dev/null @@ -1,82 +0,0 @@ -import { - normalizeScheduledAutomationRunPageLimit, - runScheduledAutomationRulePage, -} from "@/lib/automations/scheduler"; -import { parseAutomationScheduledAtMillis } from "@/lib/automations/scheduled-at"; -import { ensureUpstashSignature } from "@/lib/upstash"; -import { createSmartRouteHandler } from "@/route-handlers/smart-route-handler"; -import { yupBoolean, yupNumber, yupObject, yupString, yupTuple } from "@hexclave/shared/dist/schema-fields"; - -export const POST = createSmartRouteHandler({ - metadata: { - summary: "Run one scheduled automation rule page", - description: "Receives QStash work for a single tenancy/rule automation run page.", - tags: ["Automations"], - hidden: true, - }, - request: yupObject({ - headers: yupObject({ - "upstash-signature": yupTuple([yupString().defined()]).defined(), - }).defined(), - method: yupString().oneOf(["POST"]).defined(), - body: yupObject({ - tenancyId: yupString().defined(), - ruleId: yupString().defined(), - cursor: yupString().nullable().optional(), - limit: yupNumber().integer().min(1).max(100).optional(), - scheduledAtMillis: yupNumber().integer().optional(), - }).defined(), - }), - response: yupObject({ - statusCode: yupNumber().oneOf([200]).defined(), - bodyType: yupString().oneOf(["json"]).defined(), - body: yupObject({ - ok: yupBoolean().defined(), - status: yupString().oneOf(["ran", "skipped"]).defined(), - skipped_reason: yupString().oneOf(["tenancy-not-found", "rule-not-found", "rule-disabled"]).optional(), - evaluated_count: yupNumber().integer().optional(), - sent_count: yupNumber().integer().optional(), - suppressed_count: yupNumber().integer().optional(), - next_cursor: yupString().nullable().optional(), - enqueued_continuation: yupBoolean().optional(), - }).defined(), - }), - handler: async ({ body }, fullReq) => { - await ensureUpstashSignature(fullReq); - const scheduledAt = parseAutomationScheduledAtMillis(body.scheduledAtMillis, "scheduledAtMillis"); - const result = await runScheduledAutomationRulePage({ - tenancyId: body.tenancyId, - ruleId: body.ruleId, - cursor: body.cursor, - limit: normalizeScheduledAutomationRunPageLimit(body.limit), - scheduledAt, - now: new Date(), - }); - - if (result.status === "skipped") { - return { - statusCode: 200, - bodyType: "json", - body: { - ok: true, - status: result.status, - skipped_reason: result.reason, - }, - }; - } - - return { - statusCode: 200, - bodyType: "json", - body: { - ok: true, - status: result.status, - evaluated_count: result.result.evaluatedCount, - sent_count: result.result.sentCount, - suppressed_count: result.result.suppressedCount, - next_cursor: result.result.nextCursor, - enqueued_continuation: result.enqueuedContinuation, - }, - }; - }, -}); diff --git a/apps/backend/src/lib/automations/scheduler-state.test.ts b/apps/backend/src/lib/automations/scheduler-state.test.ts new file mode 100644 index 000000000..646ee0221 --- /dev/null +++ b/apps/backend/src/lib/automations/scheduler-state.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it } from "vitest"; +import { + automationSchedulerStateKey, + createPrismaAutomationSchedulerStateStore, + type AutomationSchedulerCheckpoint, +} from "./scheduler-state"; + +const emptyCheckpoint: AutomationSchedulerCheckpoint = { + completedTenancyCursor: null, + activeTenancyId: null, + completedRuleCursor: null, + activeRuleId: null, + nextSubjectCursor: null, +}; + +function createFakePrisma(options: { missing?: boolean } = {}) { + let state = options.missing ? null : { + key: automationSchedulerStateKey, + ...emptyCheckpoint, + leaseOwner: null as string | null, + leaseExpiresAt: null as Date | null, + }; + + const matchesWhere = (where: any) => { + const currentState = state; + if (currentState === null || where.key !== currentState.key) return false; + if (where.leaseOwner !== undefined && where.leaseOwner !== currentState.leaseOwner) return false; + if (where.leaseExpiresAt?.gt !== undefined && !(currentState.leaseExpiresAt !== null && currentState.leaseExpiresAt > where.leaseExpiresAt.gt)) return false; + if (where.OR !== undefined) { + return where.OR.some((condition: any) => { + if (condition.leaseOwner === null) return currentState.leaseOwner === null; + if (condition.leaseExpiresAt?.lte !== undefined) { + return currentState.leaseExpiresAt !== null && currentState.leaseExpiresAt <= condition.leaseExpiresAt.lte; + } + return false; + }); + } + return true; + }; + + return { + prisma: { + automationSchedulerState: { + async updateMany({ where, data }: any) { + if (!matchesWhere(where)) return { count: 0 }; + state = { ...state!, ...data }; + return { count: 1 }; + }, + async findUnique() { + return state === null ? null : { ...state }; + }, + }, + } as any, + getState: () => state, + }; +} + +function createClock(initial: string) { + let current = new Date(initial); + return { + now: () => new Date(current), + advance: (milliseconds: number) => { + current = new Date(current.getTime() + milliseconds); + }, + }; +} + +describe("Prisma automation scheduler state store", () => { + it("allows only one concurrent lease acquisition", async () => { + const fake = createFakePrisma(); + const clock = createClock("2026-07-17T12:00:00.000Z"); + let ownerSequence = 0; + const store = createPrismaAutomationSchedulerStateStore({ + prisma: fake.prisma, + now: clock.now, + ownerId: () => `00000000-0000-4000-8000-${String(++ownerSequence).padStart(12, "0")}`, + }); + + const [first, second] = await Promise.all([store.acquire(), store.acquire()]); + + expect([first, second].filter((lease) => lease !== null)).toHaveLength(1); + expect([first, second].filter((lease) => lease === null)).toHaveLength(1); + }); + + it("reclaims an expired lease and fences the stale owner", async () => { + const fake = createFakePrisma(); + const clock = createClock("2026-07-17T12:00:00.000Z"); + const firstStore = createPrismaAutomationSchedulerStateStore({ + prisma: fake.prisma, + now: clock.now, + ownerId: () => "00000000-0000-4000-8000-000000000001", + leaseDurationMs: 1_000, + renewalThresholdMs: 250, + }); + const firstLease = await firstStore.acquire() ?? undefined; + expect(firstLease).toBeDefined(); + + clock.advance(1_001); + const secondStore = createPrismaAutomationSchedulerStateStore({ + prisma: fake.prisma, + now: clock.now, + ownerId: () => "00000000-0000-4000-8000-000000000002", + leaseDurationMs: 1_000, + renewalThresholdMs: 250, + }); + const secondLease = await secondStore.acquire() ?? undefined; + expect(secondLease?.ownerId).toBe("00000000-0000-4000-8000-000000000002"); + + await expect(firstLease!.saveCheckpoint({ + ...emptyCheckpoint, + activeTenancyId: "00000000-0000-4000-8000-000000000010", + })).rejects.toThrow("lease was lost or expired"); + await expect(firstLease!.renewIfNeeded()).rejects.toThrow("lease was lost or expired"); + await expect(firstLease!.release()).rejects.toThrow("lease was lost or expired"); + + await secondLease!.saveCheckpoint({ + ...emptyCheckpoint, + activeTenancyId: "00000000-0000-4000-8000-000000000020", + }); + expect(fake.getState()?.activeTenancyId).toBe("00000000-0000-4000-8000-000000000020"); + await secondLease!.release(); + expect(fake.getState()?.leaseOwner).toBeNull(); + }); + + it("renews with injected duration values when the threshold is reached", async () => { + const fake = createFakePrisma(); + const clock = createClock("2026-07-17T12:00:00.000Z"); + const lease = await createPrismaAutomationSchedulerStateStore({ + prisma: fake.prisma, + now: clock.now, + ownerId: () => "00000000-0000-4000-8000-000000000001", + leaseDurationMs: 120_000, + renewalThresholdMs: 30_000, + }).acquire(); + + clock.advance(89_999); + await lease!.renewIfNeeded(); + expect(fake.getState()?.leaseExpiresAt).toEqual(new Date("2026-07-17T12:02:00.000Z")); + + clock.advance(2); + await lease!.renewIfNeeded(); + expect(fake.getState()?.leaseExpiresAt).toEqual(new Date("2026-07-17T12:03:30.001Z")); + }); + + it("fails loudly when the migration-seeded singleton is missing", async () => { + const fake = createFakePrisma({ missing: true }); + const store = createPrismaAutomationSchedulerStateStore({ + prisma: fake.prisma, + ownerId: () => "00000000-0000-4000-8000-000000000001", + }); + + await expect(store.acquire()).rejects.toThrow("Apply the scheduler-state migration"); + }); +}); diff --git a/apps/backend/src/lib/automations/scheduler-state.ts b/apps/backend/src/lib/automations/scheduler-state.ts new file mode 100644 index 000000000..54a1bfa7c --- /dev/null +++ b/apps/backend/src/lib/automations/scheduler-state.ts @@ -0,0 +1,176 @@ +import { randomUUID } from "crypto"; +import type { PrismaClientTransaction } from "@/prisma-client"; + +export const automationSchedulerStateKey = "usage-email-v1"; +export const automationSchedulerLeaseDurationMs = 120_000; +export const automationSchedulerLeaseRenewalThresholdMs = 30_000; + +export type AutomationSchedulerCheckpoint = { + completedTenancyCursor: string | null, + activeTenancyId: string | null, + completedRuleCursor: string | null, + activeRuleId: string | null, + nextSubjectCursor: string | null, +}; + +export type AutomationSchedulerLease = { + ownerId: string, + checkpoint: AutomationSchedulerCheckpoint, + renewIfNeeded: () => Promise, + saveCheckpoint: (checkpoint: AutomationSchedulerCheckpoint) => Promise, + release: () => Promise, +}; + +type AutomationSchedulerStatePrisma = Pick; + +export function createPrismaAutomationSchedulerStateStore(options: { + prisma: AutomationSchedulerStatePrisma, + now?: () => Date, + ownerId?: () => string, + leaseDurationMs?: number, + renewalThresholdMs?: number, +}) { + const now = options.now ?? (() => new Date()); + const ownerId = options.ownerId ?? randomUUID; + const leaseDurationMs = options.leaseDurationMs ?? automationSchedulerLeaseDurationMs; + const renewalThresholdMs = options.renewalThresholdMs ?? automationSchedulerLeaseRenewalThresholdMs; + validateLeaseDurations(leaseDurationMs, renewalThresholdMs); + + return { + async acquire(): Promise { + const acquiredAt = now(); + const owner = ownerId(); + const leaseExpiresAt = new Date(acquiredAt.getTime() + leaseDurationMs); + const acquired = await options.prisma.automationSchedulerState.updateMany({ + where: { + key: automationSchedulerStateKey, + OR: [ + { leaseOwner: null }, + { leaseExpiresAt: { lte: acquiredAt } }, + ], + }, + data: { + leaseOwner: owner, + leaseExpiresAt, + }, + }); + if (acquired.count === 0) { + const existing = await options.prisma.automationSchedulerState.findUnique({ + where: { key: automationSchedulerStateKey }, + select: { key: true }, + }); + if (existing === null) { + throw new Error(`Automation scheduler state "${automationSchedulerStateKey}" is missing. Apply the scheduler-state migration before running cron.`); + } + return null; + } + assertSingleMutation(acquired.count, "acquire automation scheduler lease"); + + const state = await options.prisma.automationSchedulerState.findUnique({ + where: { key: automationSchedulerStateKey }, + select: { + completedTenancyCursor: true, + activeTenancyId: true, + completedRuleCursor: true, + activeRuleId: true, + nextSubjectCursor: true, + leaseOwner: true, + leaseExpiresAt: true, + }, + }); + if (state === null || state.leaseOwner !== owner || state.leaseExpiresAt === null) { + throw new Error("Automation scheduler lease changed while it was being acquired."); + } + + let currentLeaseExpiresAt = state.leaseExpiresAt; + let checkpoint = readCheckpoint(state); + let released = false; + + const assertActive = () => { + if (released) { + throw new Error("Automation scheduler lease has already been released."); + } + }; + + const mutateOwnedUnexpiredLease = async ( + action: string, + data: Parameters[0]["data"], + ) => { + assertActive(); + const mutationNow = now(); + const result = await options.prisma.automationSchedulerState.updateMany({ + where: { + key: automationSchedulerStateKey, + leaseOwner: owner, + leaseExpiresAt: { gt: mutationNow }, + }, + data, + }); + assertSingleMutation(result.count, action); + }; + + return { + ownerId: owner, + get checkpoint() { + return checkpoint; + }, + async renewIfNeeded() { + assertActive(); + const renewalNow = now(); + if (currentLeaseExpiresAt.getTime() - renewalNow.getTime() >= renewalThresholdMs) { + return; + } + const renewedUntil = new Date(renewalNow.getTime() + leaseDurationMs); + await mutateOwnedUnexpiredLease("renew automation scheduler lease", { + leaseExpiresAt: renewedUntil, + }); + currentLeaseExpiresAt = renewedUntil; + }, + async saveCheckpoint(nextCheckpoint) { + await mutateOwnedUnexpiredLease("save automation scheduler checkpoint", nextCheckpoint); + checkpoint = nextCheckpoint; + }, + async release() { + assertActive(); + const result = await options.prisma.automationSchedulerState.updateMany({ + where: { + key: automationSchedulerStateKey, + leaseOwner: owner, + }, + data: { + leaseOwner: null, + leaseExpiresAt: null, + }, + }); + assertSingleMutation(result.count, "release automation scheduler lease"); + released = true; + }, + }; + }, + }; +} + +function readCheckpoint(state: AutomationSchedulerCheckpoint): AutomationSchedulerCheckpoint { + return { + completedTenancyCursor: state.completedTenancyCursor, + activeTenancyId: state.activeTenancyId, + completedRuleCursor: state.completedRuleCursor, + activeRuleId: state.activeRuleId, + nextSubjectCursor: state.nextSubjectCursor, + }; +} + +function validateLeaseDurations(leaseDurationMs: number, renewalThresholdMs: number) { + if (!Number.isFinite(leaseDurationMs) || leaseDurationMs <= 0) { + throw new Error("Automation scheduler lease duration must be positive."); + } + if (!Number.isFinite(renewalThresholdMs) || renewalThresholdMs <= 0 || renewalThresholdMs >= leaseDurationMs) { + throw new Error("Automation scheduler renewal threshold must be positive and shorter than the lease duration."); + } +} + +function assertSingleMutation(count: number, action: string) { + if (count !== 1) { + throw new Error(`Unable to ${action}; the scheduler lease was lost or expired.`); + } +} diff --git a/apps/backend/src/lib/automations/scheduler.test.ts b/apps/backend/src/lib/automations/scheduler.test.ts index 1b365f0e1..bbaa755d9 100644 --- a/apps/backend/src/lib/automations/scheduler.test.ts +++ b/apps/backend/src/lib/automations/scheduler.test.ts @@ -1,124 +1,120 @@ import { describe, expect, it, vi } from "vitest"; +import type { AutomationSchedulerCheckpoint, AutomationSchedulerLease } from "./scheduler-state"; const captureErrorMock = vi.hoisted(() => vi.fn()); vi.mock("@/lib/automations/actions/send-email", () => ({ createSendEmailActionAdapter: () => ({}), })); - vi.mock("@/lib/automations/execution-state-store", () => ({ createPrismaAutomationRuleExecutionStateStore: () => ({}), })); - vi.mock("@/lib/automations/sources/payments-item-quota", () => ({ createPaymentsItemQuotaSourceAdapter: () => ({}), paymentsItemQuotaCustomerDataReaders: {}, prismaPaymentsItemQuotaProjectUserReader: {}, })); - -vi.mock("@/lib/emails", () => ({ - sendEmailToMany: async () => {}, -})); - -vi.mock("@/lib/tenancies", () => ({ - getTenancy: async () => null, -})); - +vi.mock("@/lib/emails", () => ({ sendEmailToMany: async () => {} })); +vi.mock("@/lib/tenancies", () => ({ getTenancy: async () => null })); vi.mock("@/prisma-client", () => ({ getPrismaClientForTenancy: async () => ({}), globalPrismaClient: { - tenancy: { - findMany: async () => [], - }, - outgoingRequest: { - createMany: async () => ({ count: 0 }), - }, + tenancy: { findMany: async () => [] }, + automationSchedulerState: {}, }, })); - vi.mock("@hexclave/shared/dist/utils/errors", async (importOriginal) => { const actual = await importOriginal(); - return { - ...actual, - captureError: captureErrorMock, - }; + return { ...actual, captureError: captureErrorMock }; }); import { - discoverEnabledScheduledAutomationRules, - enqueueScheduledAutomationRuns, - getScheduledAutomationDeduplicationKey, - runScheduledAutomationRulePage, - scheduledAutomationRunRoutePath, + normalizeScheduledAutomationDiscoveryLimit, + normalizeScheduledAutomationMaxPages, + normalizeScheduledAutomationRunPageLimit, + normalizeScheduledAutomationWorkBudgetMs, + runScheduledAutomations, } from "./scheduler"; -import { AutomationRunResult } from "./run-route"; +import type { AutomationRunResult } from "./run-route"; -const enabledRuleId = "low-api-credits"; +const ruleId = "low-api-credits"; -function createAutomationRule(options: { - enabled?: boolean, - sourceType?: string, -} = {}) { +function emptyCheckpoint(): AutomationSchedulerCheckpoint { + return { + completedTenancyCursor: null, + activeTenancyId: null, + completedRuleCursor: null, + activeRuleId: null, + nextSubjectCursor: null, + }; +} + +function createStateHarness(initial: AutomationSchedulerCheckpoint = emptyCheckpoint()) { + let checkpoint = initial; + const saveCheckpoint = vi.fn(async (next: AutomationSchedulerCheckpoint) => { + checkpoint = next; + }); + const release = vi.fn(async () => {}); + const renewIfNeeded = vi.fn(async () => {}); + const acquire = vi.fn(async (): Promise => ({ + ownerId: "00000000-0000-4000-8000-000000000001", + get checkpoint() { + return checkpoint; + }, + saveCheckpoint, + release, + renewIfNeeded, + })); + return { + stateStore: { acquire }, + getCheckpoint: () => checkpoint, + acquire, + saveCheckpoint, + release, + renewIfNeeded, + }; +} + +function createPrisma(tenancyIds: string[]) { + return { + tenancy: { + findMany: vi.fn(async (options: any) => tenancyIds + .filter((id) => options.where.id?.gt === undefined || id > options.where.id.gt) + .slice(0, options.take) + .map((id) => ({ id }))), + }, + }; +} + +function createRule(options: { enabled?: boolean, sourceType?: string } = {}) { return { enabled: options.enabled ?? true, source: { type: options.sourceType ?? "payments-item-quota", - itemId: "api_credits", + itemId: "api-credits", customerType: "user", - thresholds: { - nearRemainingQuantity: 10, - }, + thresholds: { nearRemainingQuantity: 10 }, }, action: { type: "send-email", templateId: "8c6f6960-7a87-4ebd-b2a6-bfd06d68e2d1", notificationCategoryName: "Marketing", }, - cooldown: { - days: 7, - }, + cooldown: { days: 7 }, }; } -function createTenancy(options: { - enabled?: boolean, - sourceType?: string, - rules?: Record>, -} = {}) { +function createTenancy(id: string, rules: Record> = { [ruleId]: createRule() }) { return { - id: "tenancy-1", - branchId: "main", - organization: null, - project: { - id: "project-1", - display_name: "Acme App", - description: "", - logo_url: null, - logo_full_url: null, - logo_dark_mode_url: null, - logo_full_dark_mode_url: null, - created_at_millis: 0, - is_production_mode: false, - is_development_environment: false, - owner_team_id: null, - onboarding_status: "completed", - onboarding_state: undefined, - pushed_config_error: null, - config_warnings: [], - }, - config: { - automations: { - rules: options.rules ?? { - [enabledRuleId]: createAutomationRule(options), - }, - }, - }, + id, + project: { display_name: "Acme App" }, + config: { automations: { rules } }, }; } -function createRunResult(nextCursor: string | null): AutomationRunResult { +function createRunResult(nextCursor: string | null, options: Partial = {}): AutomationRunResult { return { - ruleId: enabledRuleId, + ruleId, mode: "run", evaluatedCount: 100, eligibleCount: 1, @@ -126,258 +122,248 @@ function createRunResult(nextCursor: string | null): AutomationRunResult { sentCount: 1, nextCursor, decisions: [], + ...options, }; } -describe("scheduled automation discovery", () => { - it("finds enabled V1 rules from bounded tenancy pages", async () => { - const prisma = { - tenancy: { - findMany: vi.fn(async () => [{ id: "tenancy-1" }, { id: "tenancy-2" }]), - }, - }; - const getTenancyById = vi.fn(async (tenancyId: string) => tenancyId === "tenancy-1" - ? createTenancy() - : createTenancy({ enabled: false })); - - await expect(discoverEnabledScheduledAutomationRules({ - prisma, - getTenancyById, - limit: 2, - cursor: "previous-tenancy", - })).resolves.toMatchInlineSnapshot(` - { - "nextCursor": "tenancy-2", - "scannedTenancyCount": 2, - "targets": [ - { - "ruleId": "low-api-credits", - "tenancyId": "tenancy-1", - }, - ], - } - `); - expect(prisma.tenancy.findMany).toHaveBeenCalledWith({ - where: { - id: { - gt: "previous-tenancy", - }, - }, - orderBy: { - id: "asc", - }, - take: 2, - select: { - id: true, - }, - }); - }); - - it("reports malformed enabled rules and continues discovering later valid rules", async () => { - captureErrorMock.mockClear(); - const prisma = { - tenancy: { - findMany: vi.fn(async () => [{ id: "tenancy-1" }]), - }, - }; - - await expect(discoverEnabledScheduledAutomationRules({ - prisma, - getTenancyById: async () => createTenancy({ - rules: { - invalidRule: createAutomationRule({ sourceType: "client-push-quota" }), - [enabledRuleId]: createAutomationRule(), - }, - }), - })).resolves.toMatchInlineSnapshot(` - { - "nextCursor": null, - "scannedTenancyCount": 1, - "targets": [ - { - "ruleId": "low-api-credits", - "tenancyId": "tenancy-1", - }, - ], - } - `); - expect(captureErrorMock).toHaveBeenCalledWith("automation-scheduler-invalid-rule", expect.any(Error)); +describe("scheduled automation bounds", () => { + it("strictly normalizes internal limits", () => { + expect(normalizeScheduledAutomationDiscoveryLimit(900)).toBe(500); + expect(normalizeScheduledAutomationRunPageLimit(900)).toBe(100); + expect(normalizeScheduledAutomationMaxPages(900)).toBe(10); + expect(normalizeScheduledAutomationWorkBudgetMs(90_000)).toBe(45_000); }); }); -describe("scheduled automation queueing", () => { - it("enqueues deduped QStash work through OutgoingRequest", async () => { - const prisma = { - outgoingRequest: { - createMany: vi.fn(async () => ({ count: 1 })), - }, - }; - - await expect(enqueueScheduledAutomationRuns({ - prisma, - targets: [{ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - }], - scheduledAt: new Date("2026-07-01T12:00:00.000Z"), - limit: 75, - })).resolves.toEqual({ enqueuedCount: 1 }); - - expect(prisma.outgoingRequest.createMany).toHaveBeenCalledWith({ - data: [{ - deduplicationKey: "automation-rule-run:tenancy-1:low-api-credits:start", - qstashOptions: { - body: { - cursor: null, - limit: 75, - ruleId: "low-api-credits", - scheduledAtMillis: 1782907200000, - tenancyId: "tenancy-1", - }, - flowControl: { - key: "automation-rule-run:tenancy-1", - parallelism: 1, - }, - url: scheduledAutomationRunRoutePath, - }, - }], - skipDuplicates: true, +describe("native cron automation traversal", () => { + it("returns without work when another invocation holds the lease", async () => { + await expect(runScheduledAutomations({ + stateStore: { acquire: async () => null }, + })).resolves.toMatchObject({ + status: "lease-held", + pagesProcessed: 0, + sentCount: 0, }); }); - it("reports zero enqueues when a duplicate pending OutgoingRequest already exists", async () => { - const prisma = { - outgoingRequest: { - createMany: vi.fn(async () => ({ count: 0 })), - }, - }; + it("persists the next subject cursor only after a page succeeds", async () => { + const state = createStateHarness(); + const runRule = vi.fn(async () => createRunResult("00000000-0000-4000-8000-000000000100")); - await expect(enqueueScheduledAutomationRuns({ - prisma, - targets: [{ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - }], - scheduledAt: new Date("2026-07-01T12:00:00.000Z"), - })).resolves.toEqual({ enqueuedCount: 0 }); - - expect(prisma.outgoingRequest.createMany).toHaveBeenCalledWith(expect.objectContaining({ - skipDuplicates: true, - })); - }); - - it("uses cursor-specific dedupe keys for continuation pages", () => { - expect(getScheduledAutomationDeduplicationKey({ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - cursor: "user-100", - })).toBe("automation-rule-run:tenancy-1:low-api-credits:user-100"); - }); -}); - -describe("scheduled automation worker orchestration", () => { - it("runs a page and enqueues continuation when the evaluator returns a next cursor", async () => { - const runRule = vi.fn(async () => createRunResult("user-100")); - const enqueueContinuation = vi.fn(async () => ({ enqueuedCount: 1 })); - - await expect(runScheduledAutomationRulePage({ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - limit: 100, - scheduledAt: new Date("2026-07-01T12:00:00.000Z"), - now: new Date("2026-07-01T12:01:00.000Z"), - getTenancyById: async () => createTenancy(), + await expect(runScheduledAutomations({ + prisma: createPrisma(["00000000-0000-4000-8000-000000000010"]), + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id), runRule, - enqueueContinuation, - })).resolves.toMatchInlineSnapshot(` - { - "enqueuedContinuation": true, - "result": { - "decisions": [], - "eligibleCount": 1, - "evaluatedCount": 100, - "mode": "run", - "nextCursor": "user-100", - "ruleId": "low-api-credits", - "sentCount": 1, - "suppressedCount": 0, - }, - "status": "ran", - } - `); + maxPages: 1, + })).resolves.toMatchObject({ + status: "ran", + pagesProcessed: 1, + evaluatedCount: 100, + sentCount: 1, + }); expect(runRule).toHaveBeenCalledWith(expect.objectContaining({ cursor: null, limit: 100, - ruleId: enabledRuleId, })); - expect(enqueueContinuation).toHaveBeenCalledWith({ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - cursor: "user-100", - limit: 100, - scheduledAt: new Date("2026-07-01T12:00:00.000Z"), + expect(state.getCheckpoint()).toMatchObject({ + activeRuleId: ruleId, + nextSubjectCursor: "00000000-0000-4000-8000-000000000100", }); + expect(state.renewIfNeeded).toHaveBeenCalledOnce(); + expect(state.release).toHaveBeenCalledOnce(); }); - it("reports duplicate continuation work without failing the completed page", async () => { - const runRule = vi.fn(async () => createRunResult("user-100")); - const enqueueContinuation = vi.fn(async () => ({ enqueuedCount: 0 })); + it("uses a monotonic clock for the work budget", async () => { + const state = createStateHarness(); + let elapsedMillis = 0; + let wallClockMillis = Date.UTC(2026, 0, 1); + const runRule = vi.fn(async () => { + elapsedMillis = 10; + return createRunResult("00000000-0000-4000-8000-000000000100"); + }); - await expect(runScheduledAutomationRulePage({ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - limit: 100, - scheduledAt: new Date("2026-07-01T12:00:00.000Z"), - now: new Date("2026-07-01T12:01:00.000Z"), - getTenancyById: async () => createTenancy(), + await expect(runScheduledAutomations({ + prisma: createPrisma(["00000000-0000-4000-8000-000000000010"]), + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id), runRule, - enqueueContinuation, + now: () => new Date(wallClockMillis -= 60_000), + elapsedNow: () => elapsedMillis, + workBudgetMs: 10, })).resolves.toMatchObject({ - status: "ran", - enqueuedContinuation: false, - result: { - nextCursor: "user-100", - }, + pagesProcessed: 1, }); - }); - - it("propagates continuation enqueue failures so the queue can retry the page", async () => { - const runRule = vi.fn(async () => createRunResult("user-100")); - const enqueueContinuation = vi.fn(async () => { - throw new Error("queue unavailable"); - }); - - await expect(runScheduledAutomationRulePage({ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - limit: 100, - scheduledAt: new Date("2026-07-01T12:00:00.000Z"), - now: new Date("2026-07-01T12:01:00.000Z"), - getTenancyById: async () => createTenancy(), - runRule, - enqueueContinuation, - })).rejects.toThrow("queue unavailable"); expect(runRule).toHaveBeenCalledOnce(); + expect(state.getCheckpoint()).toMatchObject({ + activeRuleId: ruleId, + nextSubjectCursor: "00000000-0000-4000-8000-000000000100", + }); }); - it("skips stale queued work when the rule was disabled after enqueue", async () => { + it("preserves the current page checkpoint and fails on transient execution errors", async () => { + const initial = { + completedTenancyCursor: "00000000-0000-4000-8000-000000000001", + activeTenancyId: "00000000-0000-4000-8000-000000000010", + completedRuleCursor: null, + activeRuleId: ruleId, + nextSubjectCursor: "00000000-0000-4000-8000-000000000100", + }; + const state = createStateHarness(initial); + const runRule = vi.fn(async () => { + throw new Error("email enqueue failed"); + }); + + await expect(runScheduledAutomations({ + prisma: createPrisma([]), + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id), + runRule, + })).rejects.toThrow("email enqueue failed"); + + expect(runRule).toHaveBeenCalledWith(expect.objectContaining({ + cursor: initial.nextSubjectCursor, + })); + expect(state.getCheckpoint()).toEqual(initial); + expect(state.release).toHaveBeenCalledOnce(); + }); + + it("propagates discovery database failures without changing the checkpoint", async () => { + const state = createStateHarness(); + const prisma = { + tenancy: { + findMany: vi.fn(async () => { + throw new Error("database unavailable"); + }), + }, + }; + + await expect(runScheduledAutomations({ + prisma, + stateStore: state.stateStore, + })).rejects.toThrow("database unavailable"); + + expect(state.getCheckpoint()).toEqual(emptyCheckpoint()); + expect(state.release).toHaveBeenCalledOnce(); + }); + + it("advances a deleted tenancy without executing a rule", async () => { + const state = createStateHarness(); + const tenancyId = "00000000-0000-4000-8000-000000000010"; const runRule = vi.fn(async () => createRunResult(null)); - await expect(runScheduledAutomationRulePage({ - tenancyId: "tenancy-1", - ruleId: enabledRuleId, - scheduledAt: new Date("2026-07-01T12:00:00.000Z"), - now: new Date("2026-07-01T12:01:00.000Z"), - getTenancyById: async () => createTenancy({ enabled: false }), + await expect(runScheduledAutomations({ + prisma: createPrisma([tenancyId]), + stateStore: state.stateStore, + getTenancyById: async () => null, runRule, - })).resolves.toMatchInlineSnapshot(` - { - "reason": "rule-disabled", - "status": "skipped", - } - `); + maxTenancies: 1, + })).resolves.toMatchObject({ + tenanciesScanned: 1, + cycleCompleted: false, + }); + + expect(state.getCheckpoint()).toEqual({ + completedTenancyCursor: tenancyId, + activeTenancyId: null, + completedRuleCursor: null, + activeRuleId: null, + nextSubjectCursor: null, + }); + expect(runRule).not.toHaveBeenCalled(); + }); + + it("completes rule and tenancy boundaries without resetting the full cycle", async () => { + const state = createStateHarness(); + const tenancyId = "00000000-0000-4000-8000-000000000010"; + + const first = await runScheduledAutomations({ + prisma: createPrisma([tenancyId]), + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id), + runRule: async () => createRunResult(null), + maxPages: 1, + }); + expect(first.cycleCompleted).toBe(false); + expect(state.getCheckpoint()).toEqual({ + completedTenancyCursor: null, + activeTenancyId: tenancyId, + completedRuleCursor: ruleId, + activeRuleId: null, + nextSubjectCursor: null, + }); + + const second = await runScheduledAutomations({ + prisma: createPrisma([tenancyId]), + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id), + runRule: async () => createRunResult(null), + }); + expect(second.cycleCompleted).toBe(true); + expect(state.getCheckpoint()).toEqual(emptyCheckpoint()); + }); + + it("continues after a full discovery page and resets only at the actual end", async () => { + const tenancyIds = Array.from({ length: 501 }, (_, index) => ( + `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}` + )); + const state = createStateHarness(); + const prisma = createPrisma(tenancyIds); + + const first = await runScheduledAutomations({ + prisma, + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id, {}), + runRule: async () => createRunResult(null), + maxTenancies: 500, + }); + expect(first).toMatchObject({ tenanciesScanned: 500, cycleCompleted: false }); + expect(state.getCheckpoint().completedTenancyCursor).toBe(tenancyIds[499]); + + const second = await runScheduledAutomations({ + prisma, + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id, {}), + runRule: async () => createRunResult(null), + maxTenancies: 500, + }); + expect(second).toMatchObject({ tenanciesScanned: 1, cycleCompleted: true }); + expect(state.getCheckpoint()).toEqual(emptyCheckpoint()); + }); + + it("reports malformed rules, advances them, and runs the next valid rule", async () => { + captureErrorMock.mockClear(); + const state = createStateHarness(); + const validRuleId = "valid-rule"; + const runRule = vi.fn(async () => createRunResult(null, { ruleId: validRuleId })); + + await runScheduledAutomations({ + prisma: createPrisma(["00000000-0000-4000-8000-000000000010"]), + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id, { + "invalid-rule": createRule({ sourceType: "client-push-quota" }), + [validRuleId]: createRule(), + }), + runRule, + maxPages: 1, + }); + + expect(captureErrorMock).toHaveBeenCalledWith("automation-scheduler-invalid-rule", expect.any(Error)); + expect(runRule).toHaveBeenCalledWith(expect.objectContaining({ ruleId: validRuleId })); + }); + + it("skips disabled rules without executing them", async () => { + const state = createStateHarness(); + const runRule = vi.fn(async () => createRunResult(null)); + + await runScheduledAutomations({ + prisma: createPrisma(["00000000-0000-4000-8000-000000000010"]), + stateStore: state.stateStore, + getTenancyById: async (id) => createTenancy(id, { [ruleId]: createRule({ enabled: false }) }), + runRule, + }); expect(runRule).not.toHaveBeenCalled(); }); diff --git a/apps/backend/src/lib/automations/scheduler.ts b/apps/backend/src/lib/automations/scheduler.ts index f2e4f0dee..364bb7019 100644 --- a/apps/backend/src/lib/automations/scheduler.ts +++ b/apps/backend/src/lib/automations/scheduler.ts @@ -1,9 +1,10 @@ import { createSendEmailActionAdapter } from "@/lib/automations/actions/send-email"; import { createPrismaAutomationRuleExecutionStateStore } from "@/lib/automations/execution-state-store"; import { - AutomationRunResult, - runAutomationRuleForRoute, -} from "@/lib/automations/run-route"; + createPrismaAutomationSchedulerStateStore, + type AutomationSchedulerCheckpoint, + type AutomationSchedulerLease, +} from "@/lib/automations/scheduler-state"; import { createPaymentsItemQuotaSourceAdapter, paymentsItemQuotaCustomerDataReaders, @@ -13,82 +14,31 @@ import { sendEmailToMany } from "@/lib/emails"; import { getTenancy, type Tenancy } from "@/lib/tenancies"; import { getPrismaClientForTenancy, globalPrismaClient } from "@/prisma-client"; import { captureError, HexclaveAssertionError } from "@hexclave/shared/dist/utils/errors"; -import { assertSupportedAutomationRule, AutomationRuleTenancy, getAutomationRule, listAutomationRules } from "./rules"; +import { stringCompare } from "@hexclave/shared/dist/utils/strings"; +import { runAutomationRuleForRoute, type AutomationRunResult } from "./run-route"; +import { assertSupportedAutomationRule, type AutomationRuleTenancy, listAutomationRules } from "./rules"; export const scheduledAutomationDiscoveryLimit = 500; export const scheduledAutomationRunPageLimit = 100; -export const scheduledAutomationRunRoutePath = "/api/latest/internal/automations/scheduled-run"; - -type AutomationScheduleTarget = { - tenancyId: string, - ruleId: string, -}; - -export type AutomationScheduleDiscoveryResult = { - scannedTenancyCount: number, - targets: AutomationScheduleTarget[], - nextCursor: string | null, -}; - -export type AutomationScheduledRunPayload = { - tenancyId: string, - ruleId: string, - cursor: string | null, - limit: number, - scheduledAtMillis: number, -}; - -export type AutomationScheduledRunResult = - | { - status: "skipped", - reason: "tenancy-not-found" | "rule-not-found" | "rule-disabled", - } - | { - status: "ran", - result: AutomationRunResult, - enqueuedContinuation: boolean, - }; +export const scheduledAutomationMaxPages = 10; +export const scheduledAutomationDefaultPages = 5; +export const scheduledAutomationWorkBudgetMs = 45_000; type AutomationDiscoveryPrisma = { tenancy: { findMany: (options: { where: { - id?: { - gt: string, - }, - }, - orderBy: { - id: "asc", + id?: { gt: string }, }, + orderBy: { id: "asc" }, take: number, - select: { - id: true, - }, + select: { id: true }, }) => Promise>, }, }; -type AutomationQueuePrisma = { - outgoingRequest: { - createMany: (options: { - data: Array<{ - qstashOptions: { - url: string, - body: AutomationScheduledRunPayload, - flowControl: { - key: string, - parallelism: number, - }, - }, - deduplicationKey: string, - }>, - skipDuplicates: true, - }) => Promise<{ count: number }>, - }, -}; - -type ScheduledAutomationRunner = (options: { - tenancy: AutomationRuleTenancy, +type ScheduledAutomationRunner = (options: { + tenancy: TTenancy, ruleId: string, cursor: string | null, limit: number, @@ -96,6 +46,32 @@ type ScheduledAutomationRunner = (options: { now: Date, }) => Promise; +type AutomationSchedulerStateStore = { + acquire: () => Promise, +}; + +type ScheduledAutomationOptions = { + prisma?: AutomationDiscoveryPrisma, + stateStore?: AutomationSchedulerStateStore, + now?: () => Date, + elapsedNow?: () => number, + pageLimit?: number, + maxPages?: number, + maxTenancies?: number, + workBudgetMs?: number, +}; + +export type ScheduledAutomationCronResult = { + status: "ran" | "lease-held", + tenanciesScanned: number, + rulesProcessed: number, + pagesProcessed: number, + evaluatedCount: number, + sentCount: number, + suppressedCount: number, + cycleCompleted: boolean, +}; + export function normalizeScheduledAutomationDiscoveryLimit(limit: number | undefined) { if (limit === undefined) return scheduledAutomationDiscoveryLimit; return Math.max(1, Math.min(Math.floor(limit), scheduledAutomationDiscoveryLimit)); @@ -106,227 +82,260 @@ export function normalizeScheduledAutomationRunPageLimit(limit: number | undefin return Math.max(1, Math.min(Math.floor(limit), scheduledAutomationRunPageLimit)); } -export async function discoverEnabledScheduledAutomationRules(options: { - prisma?: AutomationDiscoveryPrisma, - getTenancyById?: (tenancyId: string) => Promise, - limit?: number, - cursor?: string | null, -} = {}): Promise { - const prisma = options.prisma ?? globalPrismaClient; - const limit = normalizeScheduledAutomationDiscoveryLimit(options.limit); - const tenancyRows = await prisma.tenancy.findMany({ - where: { - ...(options.cursor == null ? {} : { - id: { - gt: options.cursor, - }, - }), - }, - orderBy: { - id: "asc", - }, - take: limit, - select: { - id: true, - }, - }); - const getTenancyById = options.getTenancyById ?? getTenancy; - const targets: AutomationScheduleTarget[] = []; +export function normalizeScheduledAutomationMaxPages(limit: number | undefined) { + if (limit === undefined) return scheduledAutomationDefaultPages; + return Math.max(1, Math.min(Math.floor(limit), scheduledAutomationMaxPages)); +} - for (const row of tenancyRows) { - const tenancy = await getTenancyById(row.id); +export function normalizeScheduledAutomationWorkBudgetMs(value: number | undefined) { + if (value === undefined) return scheduledAutomationWorkBudgetMs; + return Math.max(1, Math.min(Math.floor(value), scheduledAutomationWorkBudgetMs)); +} + +export function runScheduledAutomations( + options?: ScheduledAutomationOptions & { + getTenancyById?: undefined, + runRule?: undefined, + }, +): Promise; +export function runScheduledAutomations( + options: ScheduledAutomationOptions & { + getTenancyById: (tenancyId: string) => Promise, + runRule: ScheduledAutomationRunner, + }, +): Promise; +export async function runScheduledAutomations(options: ScheduledAutomationOptions & { + getTenancyById?: (tenancyId: string) => Promise, + runRule?: ScheduledAutomationRunner, +} = {}): Promise { + if (options.getTenancyById === undefined && options.runRule === undefined) { + return await runScheduledAutomationsWithDependencies(options, { + getTenancyById: getTenancy, + runRule: runProductionScheduledAutomationRulePage, + }); + } + if (options.getTenancyById === undefined || options.runRule === undefined) { + throw new Error("Automation scheduler test dependencies must provide both getTenancyById and runRule."); + } + return await runScheduledAutomationsWithDependencies(options, { + getTenancyById: options.getTenancyById, + runRule: options.runRule, + }); +} + +async function runScheduledAutomationsWithDependencies( + options: ScheduledAutomationOptions, + dependencies: { + getTenancyById: (tenancyId: string) => Promise, + runRule: ScheduledAutomationRunner, + }, +): Promise { + const prisma = options.prisma ?? globalPrismaClient; + const stateStore = options.stateStore ?? createPrismaAutomationSchedulerStateStore({ + prisma: globalPrismaClient, + }); + const lease = await stateStore.acquire(); + if (lease === null) { + return emptyCronResult("lease-held"); + } + + try { + const result = await runWithLease({ + prisma, + lease, + getTenancyById: dependencies.getTenancyById, + runRule: dependencies.runRule, + now: options.now ?? (() => new Date()), + elapsedNow: options.elapsedNow ?? (() => performance.now()), + pageLimit: normalizeScheduledAutomationRunPageLimit(options.pageLimit), + maxPages: normalizeScheduledAutomationMaxPages(options.maxPages), + maxTenancies: normalizeScheduledAutomationDiscoveryLimit(options.maxTenancies), + workBudgetMs: normalizeScheduledAutomationWorkBudgetMs(options.workBudgetMs), + }); + await lease.release(); + return result; + } catch (error) { + try { + await lease.release(); + } catch (releaseError) { + captureError("automation-scheduler-release-after-failure", new HexclaveAssertionError("Failed to release the automation scheduler lease after an execution failure.", { + cause: releaseError, + })); + } + throw error; + } +} + +async function runWithLease(options: { + prisma: AutomationDiscoveryPrisma, + lease: AutomationSchedulerLease, + getTenancyById: (tenancyId: string) => Promise, + runRule: ScheduledAutomationRunner, + now: () => Date, + elapsedNow: () => number, + pageLimit: number, + maxPages: number, + maxTenancies: number, + workBudgetMs: number, +}): Promise { + const startedAt = options.now(); + const startedElapsedAt = options.elapsedNow(); + const scheduledAt = startedAt; + let checkpoint = options.lease.checkpoint; + let tenanciesScanned = 0; + let rulesProcessed = 0; + let pagesProcessed = 0; + let evaluatedCount = 0; + let sentCount = 0; + let suppressedCount = 0; + let cycleCompleted = false; + let discoveredTenancyIds: string[] = []; + + const saveCheckpoint = async (next: AutomationSchedulerCheckpoint) => { + await options.lease.saveCheckpoint(next); + checkpoint = next; + }; + + while ( + pagesProcessed < options.maxPages + && tenanciesScanned < options.maxTenancies + && options.elapsedNow() - startedElapsedAt < options.workBudgetMs + ) { + if (checkpoint.activeTenancyId === null) { + if (discoveredTenancyIds.length === 0) { + const rows = await options.prisma.tenancy.findMany({ + where: checkpoint.completedTenancyCursor === null ? {} : { + id: { gt: checkpoint.completedTenancyCursor }, + }, + orderBy: { id: "asc" }, + take: Math.min( + scheduledAutomationDiscoveryLimit, + options.maxTenancies - tenanciesScanned, + ), + select: { id: true }, + }); + if (rows.length === 0) { + await saveCheckpoint(emptyCheckpoint()); + cycleCompleted = true; + break; + } + discoveredTenancyIds = rows.map((row) => row.id); + } + + const nextTenancyId = discoveredTenancyIds.shift(); + if (nextTenancyId === undefined) { + throw new Error("Automation scheduler discovery returned no next tenancy unexpectedly."); + } + await saveCheckpoint({ + ...checkpoint, + activeTenancyId: nextTenancyId, + completedRuleCursor: null, + activeRuleId: null, + nextSubjectCursor: null, + }); + tenanciesScanned++; + } + + const activeTenancyId = checkpoint.activeTenancyId; + if (activeTenancyId === null) { + throw new Error("Automation scheduler checkpoint lost its active tenancy unexpectedly."); + } + const tenancy = await options.getTenancyById(activeTenancyId); if (tenancy === null) { + await saveCheckpoint(completeActiveTenancy(checkpoint)); continue; } - for (const { ruleId, rule } of listAutomationRules(tenancy)) { - if (!rule.enabled) { + const sortedRules = listAutomationRules(tenancy) + .sort((left, right) => stringCompare(left.ruleId, right.ruleId)); + if (checkpoint.activeRuleId === null) { + const nextRule = sortedRules.find(({ ruleId }) => ( + checkpoint.completedRuleCursor === null || stringCompare(ruleId, checkpoint.completedRuleCursor) > 0 + )); + if (nextRule === undefined) { + await saveCheckpoint(completeActiveTenancy(checkpoint)); continue; } - try { - assertSupportedAutomationRule(ruleId, rule); - } catch (error) { - captureError("automation-scheduler-invalid-rule", new HexclaveAssertionError(`Skipping invalid scheduled automation rule "${ruleId}" for tenancy "${tenancy.id}".`, { - cause: error, - tenancyId: tenancy.id, - ruleId, - })); - continue; - } - targets.push({ + await saveCheckpoint({ + ...checkpoint, + activeRuleId: nextRule.ruleId, + nextSubjectCursor: null, + }); + } + + const activeRuleId = checkpoint.activeRuleId; + if (activeRuleId === null) { + throw new Error("Automation scheduler checkpoint lost its active rule unexpectedly."); + } + const ruleEntry = sortedRules.find(({ ruleId }) => ruleId === activeRuleId); + if (ruleEntry === undefined) { + await saveCheckpoint(completeActiveRule(checkpoint, activeRuleId)); + rulesProcessed++; + continue; + } + if (!ruleEntry.rule.enabled) { + await saveCheckpoint(completeActiveRule(checkpoint, activeRuleId)); + rulesProcessed++; + continue; + } + try { + assertSupportedAutomationRule(activeRuleId, ruleEntry.rule); + } catch (error) { + captureError("automation-scheduler-invalid-rule", new HexclaveAssertionError(`Skipping invalid scheduled automation rule "${activeRuleId}" for tenancy "${tenancy.id}".`, { + cause: error, tenancyId: tenancy.id, - ruleId, + ruleId: activeRuleId, + })); + await saveCheckpoint(completeActiveRule(checkpoint, activeRuleId)); + rulesProcessed++; + continue; + } + + if ( + pagesProcessed >= options.maxPages + || options.elapsedNow() - startedElapsedAt >= options.workBudgetMs + ) { + break; + } + await options.lease.renewIfNeeded(); + + const result = await options.runRule({ + tenancy, + ruleId: activeRuleId, + cursor: checkpoint.nextSubjectCursor, + limit: options.pageLimit, + scheduledAt, + now: options.now(), + }); + pagesProcessed++; + evaluatedCount += result.evaluatedCount; + sentCount += result.sentCount; + suppressedCount += result.suppressedCount; + + if (result.nextCursor === null) { + await saveCheckpoint(completeActiveRule(checkpoint, activeRuleId)); + rulesProcessed++; + } else { + await saveCheckpoint({ + ...checkpoint, + nextSubjectCursor: result.nextCursor, }); } } - return { - scannedTenancyCount: tenancyRows.length, - targets, - nextCursor: tenancyRows.length === limit ? tenancyRows[tenancyRows.length - 1]?.id ?? null : null, - }; -} - -export async function enqueueScheduledAutomationRuns(options: { - prisma?: AutomationQueuePrisma, - targets: AutomationScheduleTarget[], - limit?: number, - cursor?: string | null, - scheduledAt: Date, -}): Promise<{ enqueuedCount: number }> { - if (options.targets.length === 0) { - return { enqueuedCount: 0 }; - } - - const prisma = options.prisma ?? globalPrismaClient; - const limit = normalizeScheduledAutomationRunPageLimit(options.limit); - const cursor = options.cursor ?? null; - const createResult = await prisma.outgoingRequest.createMany({ - data: options.targets.map((target) => ({ - qstashOptions: { - url: scheduledAutomationRunRoutePath, - body: { - tenancyId: target.tenancyId, - ruleId: target.ruleId, - cursor, - limit, - scheduledAtMillis: options.scheduledAt.getTime(), - }, - flowControl: { - key: getScheduledAutomationFlowControlKey(target.tenancyId), - parallelism: 1, - }, - }, - deduplicationKey: getScheduledAutomationDeduplicationKey({ - ...target, - cursor, - }), - })), - skipDuplicates: true, - }); - - return { - enqueuedCount: createResult.count, - }; -} - -export async function enqueueScheduledAutomationContinuation(options: { - tenancyId: string, - ruleId: string, - cursor: string, - limit: number, - scheduledAt: Date, - prisma?: AutomationQueuePrisma, -}): Promise<{ enqueuedCount: number }> { - return await enqueueScheduledAutomationRuns({ - prisma: options.prisma, - targets: [{ - tenancyId: options.tenancyId, - ruleId: options.ruleId, - }], - cursor: options.cursor, - limit: options.limit, - scheduledAt: options.scheduledAt, - }); -} - -export async function runScheduledAutomationRulePage(options: { - tenancyId: string, - ruleId: string, - cursor?: string | null, - limit?: number, - scheduledAt: Date, - now: Date, - getTenancyById?: (tenancyId: string) => Promise, - runRule?: ScheduledAutomationRunner, - enqueueContinuation?: typeof enqueueScheduledAutomationContinuation, -}): Promise { - const getTenancyById = options.getTenancyById ?? getTenancy; - const tenancy = await getTenancyById(options.tenancyId); - if (tenancy === null) { - return { - status: "skipped", - reason: "tenancy-not-found", - }; - } - - const rule = getAutomationRule(tenancy, options.ruleId); - if (rule === undefined) { - return { - status: "skipped", - reason: "rule-not-found", - }; - } - if (!rule.enabled) { - return { - status: "skipped", - reason: "rule-disabled", - }; - } - assertSupportedAutomationRule(options.ruleId, rule); - - const limit = normalizeScheduledAutomationRunPageLimit(options.limit); - const result = options.runRule === undefined - ? await runProductionScheduledAutomationRulePage({ - tenancyId: options.tenancyId, - ruleId: options.ruleId, - cursor: options.cursor ?? null, - limit, - scheduledAt: options.scheduledAt, - now: options.now, - }) - : await options.runRule({ - tenancy, - ruleId: options.ruleId, - cursor: options.cursor ?? null, - limit, - scheduledAt: options.scheduledAt, - now: options.now, - }); - - let enqueuedContinuation = false; - if (result.nextCursor !== null) { - const enqueueContinuation = options.enqueueContinuation ?? enqueueScheduledAutomationContinuation; - const enqueueResult = await enqueueContinuation({ - tenancyId: options.tenancyId, - ruleId: options.ruleId, - cursor: result.nextCursor, - limit, - scheduledAt: options.scheduledAt, - }); - enqueuedContinuation = enqueueResult.enqueuedCount > 0; - } - return { status: "ran", - result, - enqueuedContinuation, + tenanciesScanned, + rulesProcessed, + pagesProcessed, + evaluatedCount, + sentCount, + suppressedCount, + cycleCompleted, }; } async function runProductionScheduledAutomationRulePage(options: { - tenancyId: string, - ruleId: string, - cursor: string | null, - limit: number, - scheduledAt: Date, - now: Date, -}) { - const tenancy = await getTenancy(options.tenancyId); - if (tenancy === null) { - throw new Error(`Tenancy "${options.tenancyId}" disappeared between scheduled automation validation and execution.`); - } - return await runAutomationRuleForScheduledWorker({ - tenancy, - ruleId: options.ruleId, - cursor: options.cursor, - limit: options.limit, - scheduledAt: options.scheduledAt, - now: options.now, - }); -} - -async function runAutomationRuleForScheduledWorker(options: { tenancy: Tenancy, ruleId: string, cursor: string | null, @@ -334,7 +343,8 @@ async function runAutomationRuleForScheduledWorker(options: { scheduledAt: Date, now: Date, }): Promise { - const prisma = await getPrismaClientForTenancy(options.tenancy); + const tenancy = options.tenancy; + const prisma = await getPrismaClientForTenancy(tenancy); const sourceAdapter = createPaymentsItemQuotaSourceAdapter({ prisma, projectUserReader: prismaPaymentsItemQuotaProjectUserReader, @@ -342,7 +352,7 @@ async function runAutomationRuleForScheduledWorker(options: { }); return await runAutomationRuleForRoute({ - tenancy: options.tenancy, + tenancy, ruleId: options.ruleId, limit: options.limit, cursor: options.cursor, @@ -353,7 +363,7 @@ async function runAutomationRuleForScheduledWorker(options: { stateStore: createPrismaAutomationRuleExecutionStateStore(prisma), emailSender: async ({ action, scheduledAt }) => { await sendEmailToMany({ - tenancy: options.tenancy, + tenancy, recipients: [action.recipient], tsxSource: action.tsxSource, extraVariables: action.variables, @@ -369,14 +379,47 @@ async function runAutomationRuleForScheduledWorker(options: { }); } -function getScheduledAutomationFlowControlKey(tenancyId: string) { - return `automation-rule-run:${tenancyId}`; +function completeActiveRule(checkpoint: AutomationSchedulerCheckpoint, ruleId: string): AutomationSchedulerCheckpoint { + return { + ...checkpoint, + completedRuleCursor: ruleId, + activeRuleId: null, + nextSubjectCursor: null, + }; } -export function getScheduledAutomationDeduplicationKey(options: { - tenancyId: string, - ruleId: string, - cursor: string | null, -}) { - return `automation-rule-run:${options.tenancyId}:${options.ruleId}:${options.cursor ?? "start"}`; +function completeActiveTenancy(checkpoint: AutomationSchedulerCheckpoint): AutomationSchedulerCheckpoint { + if (checkpoint.activeTenancyId === null) { + throw new Error("Cannot complete an automation scheduler tenancy without an active tenancy."); + } + return { + completedTenancyCursor: checkpoint.activeTenancyId, + activeTenancyId: null, + completedRuleCursor: null, + activeRuleId: null, + nextSubjectCursor: null, + }; +} + +function emptyCheckpoint(): AutomationSchedulerCheckpoint { + return { + completedTenancyCursor: null, + activeTenancyId: null, + completedRuleCursor: null, + activeRuleId: null, + nextSubjectCursor: null, + }; +} + +function emptyCronResult(status: "lease-held"): ScheduledAutomationCronResult { + return { + status, + tenanciesScanned: 0, + rulesProcessed: 0, + pagesProcessed: 0, + evaluatedCount: 0, + sentCount: 0, + suppressedCount: 0, + cycleCompleted: false, + }; } diff --git a/apps/backend/src/lib/email-rendering.test.tsx b/apps/backend/src/lib/email-rendering.test.tsx index c78c3d11f..80716dbd6 100644 --- a/apps/backend/src/lib/email-rendering.test.tsx +++ b/apps/backend/src/lib/email-rendering.test.tsx @@ -1,4 +1,6 @@ import { describe, expect, it } from 'vitest'; +import { LightEmailTheme } from '@hexclave/shared/dist/helpers/emails'; +import { usageEmailTemplateSource } from '../../../../packages/shared/src/helpers/usage-email-template'; import { renderEmailsForTenancyBatched, renderEmailWithTemplate, type RenderEmailRequestForTenancy } from './email-rendering'; describe('renderEmailsForTenancyBatched', () => { @@ -533,6 +535,47 @@ describe('renderEmailWithTemplate', () => { } `; + it.each([ + { + thresholdKind: 'near', + currentQuantity: -1, + expectedCopy: 'Your remaining usage is low.', + unexpectedCopy: 'Your quota is at or below its limit.', + }, + { + thresholdKind: 'over', + currentQuantity: 999, + expectedCopy: 'Your quota is at or below its limit.', + unexpectedCopy: 'Your remaining usage is low.', + }, + ])('renders the built-in Usage Email from thresholdKind=$thresholdKind', async ({ + thresholdKind, + currentQuantity, + expectedCopy, + unexpectedCopy, + }) => { + const result = await renderEmailWithTemplate(usageEmailTemplateSource, LightEmailTheme, { + user: { displayName: 'QA User' }, + project: { displayName: 'QA Project' }, + variables: { + projectDisplayName: 'QA Project', + itemDisplayName: 'API requests', + currentQuantity, + thresholdKind, + }, + }); + + expect(result.status).toBe('ok'); + if (result.status === 'ok') { + expect(result.data.subject).toBe('API requests usage alert'); + expect(result.data.notificationCategory).toBe('Marketing'); + expect(result.data.html).toContain(expectedCopy); + expect(result.data.text).toContain(expectedCopy); + expect(result.data.html).not.toContain(unexpectedCopy); + expect(result.data.text).toContain(`Current API requests quantity: ${currentQuantity}`); + } + }); + it('preview mode: uses default user and project when not provided', async () => { const result = await renderEmailWithTemplate(simpleTemplate, simpleTheme, { previewMode: true, @@ -685,4 +728,3 @@ describe('renderEmailWithTemplate', () => { `); }); }); - diff --git a/packages/shared/src/config/schema.ts b/packages/shared/src/config/schema.ts index a9ee6dfae..ac0eb7de5 100644 --- a/packages/shared/src/config/schema.ts +++ b/packages/shared/src/config/schema.ts @@ -1123,6 +1123,13 @@ const organizationConfigDefaults = { }), } as const satisfies DefaultsType; +import.meta.vitest?.test("organization defaults include the built-in Usage Email template", ({ expect }) => { + expect(organizationConfigDefaults.emails.templates["28a45509-eb75-440d-b657-0a8640c775df"]).toMatchObject({ + displayName: "Usage Email", + themeId: undefined, + }); +}); + type _DeepOmitDefaultsImpl = T extends object ? ( ( & /* keys that are both in T and U, *and* the key's value in U is not a subtype of the key's value in T */ { [K in { [Ki in keyof T & keyof U]: U[Ki] extends T[Ki] ? never : Ki }[keyof T & keyof U]]: DeepOmitDefaults } diff --git a/packages/shared/src/helpers/emails.ts b/packages/shared/src/helpers/emails.ts index 084d8cd23..f143402ac 100644 --- a/packages/shared/src/helpers/emails.ts +++ b/packages/shared/src/helpers/emails.ts @@ -1,4 +1,5 @@ import { deindent } from "../utils/strings"; +import { usageEmailTemplateSource } from "./usage-email-template"; export const defaultNewTemplateSource = deindent` import { type } from "arktype" @@ -200,6 +201,7 @@ const EMAIL_TEMPLATE_TEAM_INVITATION_ID = "e84de395-2076-4831-9c19-8e9a96a868e4" const EMAIL_TEMPLATE_SIGN_IN_INVITATION_ID = "066dd73c-36da-4fd0-b6d6-ebf87683f8bc"; const EMAIL_TEMPLATE_PAYMENT_RECEIPT_ID = "70372aee-0441-4d80-974c-2e858e40123a"; const EMAIL_TEMPLATE_PAYMENT_FAILED_ID = "f64b1afe-27ec-4a28-a277-7178f3261f9a"; +const EMAIL_TEMPLATE_USAGE_EMAIL_ID = "28a45509-eb75-440d-b657-0a8640c775df"; export const DEFAULT_EMAIL_TEMPLATES = { [EMAIL_TEMPLATE_EMAIL_VERIFICATION_ID]: { @@ -236,6 +238,11 @@ export const DEFAULT_EMAIL_TEMPLATES = { "displayName": "Payment Failed", "tsxSource": "import { type } from \"arktype\";\nimport { Button, Section, Hr } from \"@react-email/components\";\nimport { Subject, NotificationCategory, Props } from \"@stackframe/emails\";\n\nexport const variablesSchema = type({\n productName: \"string\",\n amount: \"string\",\n invoiceUrl: \"string?\",\n failureReason: \"string?\"\n});\n\nexport function EmailTemplate({ user, project, variables }: Props) {\n return (\n <>\n \n \n
\n
\n

\n We couldn't process your payment\n

\n

\n Hi{user.displayName ? (\", \" + user.displayName) : \"\"}! Your payment for {variables.productName} ({variables.amount}) did not go through.\n

\n {variables.failureReason ? (\n

\n Reason: {variables.failureReason}\n

\n ) : null}\n {variables.invoiceUrl ? (\n
\n \n View invoice\n \n
\n ) : null}\n
\n
\n
\n

\n Please update your payment details to avoid service interruption.\n

\n
\n
\n \n );\n}\n\nEmailTemplate.PreviewVariables = {\n productName: \"Pro Plan\",\n amount: \"$29.00\",\n invoiceUrl: \"https://example.com/billing\",\n failureReason: \"Your card was declined\"\n} satisfies typeof variablesSchema.infer;\n", "themeId": undefined, + }, + [EMAIL_TEMPLATE_USAGE_EMAIL_ID]: { + "displayName": "Usage Email", + "tsxSource": usageEmailTemplateSource, + "themeId": undefined, } }; diff --git a/packages/shared/src/helpers/usage-email-template.ts b/packages/shared/src/helpers/usage-email-template.ts new file mode 100644 index 000000000..c652b2d74 --- /dev/null +++ b/packages/shared/src/helpers/usage-email-template.ts @@ -0,0 +1,48 @@ +export const usageEmailTemplateSource = ` + import { type } from "arktype"; + import { Heading, Section, Text } from "@react-email/components"; + import { Subject, NotificationCategory, Props } from "@stackframe/emails"; + + export const variablesSchema = type({ + projectDisplayName: "string", + itemDisplayName: "string", + currentQuantity: "number", + thresholdKind: "'near' | 'over'" + }); + + export function EmailTemplate({ variables }: Props) { + const isOverLimit = variables.thresholdKind === "over"; + + return ( + <> + + +
+ + {isOverLimit + ? variables.itemDisplayName + " limit reached" + : variables.itemDisplayName + " usage is running low"} + + + {isOverLimit + ? "Your quota is at or below its limit." + : "Your remaining usage is low."} + + + Current {variables.itemDisplayName} quantity: {variables.currentQuantity} + + + Review your plan in {variables.projectDisplayName} if you need more capacity. + +
+ + ); + } + + EmailTemplate.PreviewVariables = { + projectDisplayName: "My Project", + itemDisplayName: "API requests", + currentQuantity: 10, + thresholdKind: "near" + } satisfies typeof variablesSchema.infer; +`;