mirror of
https://github.com/stack-auth/stack.git
synced 2026-07-20 21:29:36 +08:00
Email Automation V1.5 (Fixed the Automated Email Schedule-Sender and the Email Schedule Fairness Problem)
This commit is contained in:
AddarshKS
+17
@@ -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);
|
||||
+20
@@ -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,
|
||||
}]);
|
||||
};
|
||||
@@ -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
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
};
|
||||
},
|
||||
});
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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<void>,
|
||||
saveCheckpoint: (checkpoint: AutomationSchedulerCheckpoint) => Promise<void>,
|
||||
release: () => Promise<void>,
|
||||
};
|
||||
|
||||
type AutomationSchedulerStatePrisma = Pick<PrismaClientTransaction, "automationSchedulerState">;
|
||||
|
||||
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<AutomationSchedulerLease | null> {
|
||||
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<typeof options.prisma.automationSchedulerState.updateMany>[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.`);
|
||||
}
|
||||
}
|
||||
@@ -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<typeof import("@hexclave/shared/dist/utils/errors")>();
|
||||
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<AutomationSchedulerLease> => ({
|
||||
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<string, ReturnType<typeof createAutomationRule>>,
|
||||
} = {}) {
|
||||
function createTenancy(id: string, rules: Record<string, ReturnType<typeof createRule>> = { [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> = {}): 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();
|
||||
});
|
||||
|
||||
@@ -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<Array<{ id: string }>>,
|
||||
},
|
||||
};
|
||||
|
||||
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<TTenancy extends AutomationRuleTenancy> = (options: {
|
||||
tenancy: TTenancy,
|
||||
ruleId: string,
|
||||
cursor: string | null,
|
||||
limit: number,
|
||||
@@ -96,6 +46,32 @@ type ScheduledAutomationRunner = (options: {
|
||||
now: Date,
|
||||
}) => Promise<AutomationRunResult>;
|
||||
|
||||
type AutomationSchedulerStateStore = {
|
||||
acquire: () => Promise<AutomationSchedulerLease | null>,
|
||||
};
|
||||
|
||||
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<AutomationRuleTenancy | null>,
|
||||
limit?: number,
|
||||
cursor?: string | null,
|
||||
} = {}): Promise<AutomationScheduleDiscoveryResult> {
|
||||
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<ScheduledAutomationCronResult>;
|
||||
export function runScheduledAutomations<TTenancy extends AutomationRuleTenancy>(
|
||||
options: ScheduledAutomationOptions & {
|
||||
getTenancyById: (tenancyId: string) => Promise<TTenancy | null>,
|
||||
runRule: ScheduledAutomationRunner<TTenancy>,
|
||||
},
|
||||
): Promise<ScheduledAutomationCronResult>;
|
||||
export async function runScheduledAutomations(options: ScheduledAutomationOptions & {
|
||||
getTenancyById?: (tenancyId: string) => Promise<AutomationRuleTenancy | null>,
|
||||
runRule?: ScheduledAutomationRunner<AutomationRuleTenancy>,
|
||||
} = {}): Promise<ScheduledAutomationCronResult> {
|
||||
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<TTenancy extends AutomationRuleTenancy>(
|
||||
options: ScheduledAutomationOptions,
|
||||
dependencies: {
|
||||
getTenancyById: (tenancyId: string) => Promise<TTenancy | null>,
|
||||
runRule: ScheduledAutomationRunner<TTenancy>,
|
||||
},
|
||||
): Promise<ScheduledAutomationCronResult> {
|
||||
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<TTenancy extends AutomationRuleTenancy>(options: {
|
||||
prisma: AutomationDiscoveryPrisma,
|
||||
lease: AutomationSchedulerLease,
|
||||
getTenancyById: (tenancyId: string) => Promise<TTenancy | null>,
|
||||
runRule: ScheduledAutomationRunner<TTenancy>,
|
||||
now: () => Date,
|
||||
elapsedNow: () => number,
|
||||
pageLimit: number,
|
||||
maxPages: number,
|
||||
maxTenancies: number,
|
||||
workBudgetMs: number,
|
||||
}): Promise<ScheduledAutomationCronResult> {
|
||||
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<AutomationRuleTenancy | null>,
|
||||
runRule?: ScheduledAutomationRunner,
|
||||
enqueueContinuation?: typeof enqueueScheduledAutomationContinuation,
|
||||
}): Promise<AutomationScheduledRunResult> {
|
||||
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<AutomationRunResult> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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', () => {
|
||||
`);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -1123,6 +1123,13 @@ const organizationConfigDefaults = {
|
||||
}),
|
||||
} as const satisfies DefaultsType<OrganizationRenderedConfigBeforeDefaults, [typeof environmentConfigDefaults, typeof branchConfigDefaults, typeof projectConfigDefaults]>;
|
||||
|
||||
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, U> = 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<T[K], U[K] & object> }
|
||||
|
||||
@@ -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<typeof variablesSchema.infer>) {\n return (\n <>\n <Subject value={\"Payment failed for \" + project.displayName} />\n <NotificationCategory value=\"Transactional\" />\n <div className=\"bg-white text-[#242424] font-sans text-base font-normal tracking-[0.15008px] leading-[1.5] m-0 py-8 w-full min-h-full\">\n <Section className=\"bg-white\">\n <h3 className=\"text-black font-sans font-bold text-[20px] text-center py-4 px-6 m-0\">\n We couldn't process your payment\n </h3>\n <p className=\"text-[#474849] font-sans font-normal text-[14px] text-center pt-2 px-6 pb-4 m-0\">\n Hi{user.displayName ? (\", \" + user.displayName) : \"\"}! Your payment for {variables.productName} ({variables.amount}) did not go through.\n </p>\n {variables.failureReason ? (\n <p className=\"text-[#8a8b8b] font-sans text-[12px] text-center px-6 pb-2 m-0\">\n Reason: {variables.failureReason}\n </p>\n ) : null}\n {variables.invoiceUrl ? (\n <div className=\"text-center py-3 px-6\">\n <Button\n href={variables.invoiceUrl}\n target=\"_blank\"\n className=\"text-black font-sans font-bold text-[14px] inline-block bg-[#f0f0f0] rounded-[4px] py-3 px-5 no-underline border-0\"\n >\n View invoice\n </Button>\n </div>\n ) : null}\n <div className=\"py-4 px-6\">\n <Hr />\n </div>\n <p className=\"text-[#474849] font-sans font-normal text-[12px] text-center pt-1 px-6 pb-6 m-0\">\n Please update your payment details to avoid service interruption.\n </p>\n </Section>\n </div>\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,
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -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<typeof variablesSchema.infer>) {
|
||||
const isOverLimit = variables.thresholdKind === "over";
|
||||
|
||||
return (
|
||||
<>
|
||||
<Subject value={variables.itemDisplayName + " usage alert"} />
|
||||
<NotificationCategory value="Marketing" />
|
||||
<Section>
|
||||
<Heading as="h3" className="font-sans font-bold text-[20px] text-center py-4 px-6 m-0">
|
||||
{isOverLimit
|
||||
? variables.itemDisplayName + " limit reached"
|
||||
: variables.itemDisplayName + " usage is running low"}
|
||||
</Heading>
|
||||
<Text className="font-sans font-normal text-[14px] text-center pt-2 px-6 pb-2 m-0 opacity-80">
|
||||
{isOverLimit
|
||||
? "Your quota is at or below its limit."
|
||||
: "Your remaining usage is low."}
|
||||
</Text>
|
||||
<Text className="font-sans font-normal text-[14px] text-center px-6 pb-2 m-0 opacity-80">
|
||||
Current {variables.itemDisplayName} quantity: {variables.currentQuantity}
|
||||
</Text>
|
||||
<Text className="font-sans font-normal text-[12px] text-center px-6 pb-6 m-0 opacity-60">
|
||||
Review your plan in {variables.projectDisplayName} if you need more capacity.
|
||||
</Text>
|
||||
</Section>
|
||||
</>
|
||||
);
|
||||
}
|
||||
|
||||
EmailTemplate.PreviewVariables = {
|
||||
projectDisplayName: "My Project",
|
||||
itemDisplayName: "API requests",
|
||||
currentQuantity: 10,
|
||||
thresholdKind: "near"
|
||||
} satisfies typeof variablesSchema.infer;
|
||||
`;
|
||||
Reference in New Issue
Block a user