mirror of
https://github.com/stack-auth/stack.git
synced 2026-07-20 21:29:36 +08:00
Email Automation V2 Bot Comment Fix Batch I
This commit is contained in:
AddarshKS
parent
fa2dc67165
commit
a7d6fd760d
@ -8,7 +8,7 @@ CREATE TABLE "AutomationRuleExecutionState" (
|
||||
"signalKey" TEXT NOT NULL,
|
||||
"lastTriggeredAt" TIMESTAMP(3) NOT NULL,
|
||||
"lastActionAt" TIMESTAMP(3),
|
||||
"lastEmailOutboxId" UUID,
|
||||
"emailOutboxId" UUID NOT NULL,
|
||||
"lastSourceSnapshot" JSONB NOT NULL,
|
||||
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||
|
||||
@ -40,6 +40,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
"subjectId",
|
||||
"signalKey",
|
||||
"lastTriggeredAt",
|
||||
"emailOutboxId",
|
||||
"lastSourceSnapshot",
|
||||
"createdAt",
|
||||
"updatedAt"
|
||||
@ -53,6 +54,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
${projectUserId},
|
||||
'credits:near',
|
||||
NOW(),
|
||||
${emailOutboxId}::uuid,
|
||||
${JSON.stringify(initialSnapshot)}::jsonb,
|
||||
NOW(),
|
||||
NOW()
|
||||
@ -67,7 +69,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
"subjectId",
|
||||
"signalKey",
|
||||
"lastActionAt",
|
||||
"lastEmailOutboxId",
|
||||
"emailOutboxId",
|
||||
"lastSourceSnapshot"
|
||||
FROM "AutomationRuleExecutionState"
|
||||
WHERE "tenancyId" = ${ctx.tenancyId}::uuid
|
||||
@ -83,7 +85,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
subjectId: projectUserId,
|
||||
signalKey: "credits:near",
|
||||
lastActionAt: null,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId,
|
||||
}]);
|
||||
expect(JSON.parse(inserted[0].lastSourceSnapshot)).toMatchObject(initialSnapshot);
|
||||
|
||||
@ -96,7 +98,6 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
UPDATE "AutomationRuleExecutionState"
|
||||
SET
|
||||
"lastActionAt" = NOW(),
|
||||
"lastEmailOutboxId" = ${emailOutboxId}::uuid,
|
||||
"lastSourceSnapshot" = ${JSON.stringify(updatedSnapshot)}::jsonb,
|
||||
"updatedAt" = NOW()
|
||||
WHERE "tenancyId" = ${ctx.tenancyId}::uuid
|
||||
@ -107,7 +108,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
`;
|
||||
|
||||
const updated = await sql`
|
||||
SELECT "lastActionAt", "lastEmailOutboxId", "lastSourceSnapshot"
|
||||
SELECT "lastActionAt", "emailOutboxId", "lastSourceSnapshot"
|
||||
FROM "AutomationRuleExecutionState"
|
||||
WHERE "tenancyId" = ${ctx.tenancyId}::uuid
|
||||
AND "ruleId" = 'lowCreditsUpgradeEmail'
|
||||
@ -117,7 +118,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
`;
|
||||
expect(updated).toHaveLength(1);
|
||||
expect(updated[0].lastActionAt).toBeInstanceOf(Date);
|
||||
expect(updated[0].lastEmailOutboxId).toBe(emailOutboxId);
|
||||
expect(updated[0].emailOutboxId).toBe(emailOutboxId);
|
||||
expect(JSON.parse(updated[0].lastSourceSnapshot)).toMatchObject(updatedSnapshot);
|
||||
|
||||
await expect(sql`
|
||||
@ -130,6 +131,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
"subjectId",
|
||||
"signalKey",
|
||||
"lastTriggeredAt",
|
||||
"emailOutboxId",
|
||||
"lastSourceSnapshot",
|
||||
"createdAt",
|
||||
"updatedAt"
|
||||
@ -143,6 +145,7 @@ export const postMigration = async (sql: Sql, ctx: Awaited<ReturnType<typeof pre
|
||||
${projectUserId},
|
||||
'credits:near',
|
||||
NOW(),
|
||||
${randomUUID()}::uuid,
|
||||
${JSON.stringify(initialSnapshot)}::jsonb,
|
||||
NOW(),
|
||||
NOW()
|
||||
|
||||
@ -1175,7 +1175,8 @@ model AutomationRuleExecutionState {
|
||||
signalKey String
|
||||
lastTriggeredAt DateTime
|
||||
lastActionAt DateTime?
|
||||
lastEmailOutboxId String? @db.Uuid
|
||||
// Reserved before enqueue so an ambiguous retry addresses the same EmailOutbox row.
|
||||
emailOutboxId String @db.Uuid
|
||||
lastSourceSnapshot Json
|
||||
|
||||
createdAt DateTime @default(now())
|
||||
|
||||
@ -6,8 +6,11 @@ import {
|
||||
} from "@/lib/automations/execution-state-store";
|
||||
import { AutomationSourceDecision } from "@/lib/automations/rule-evaluator";
|
||||
import {
|
||||
AutomationEmailSender,
|
||||
AutomationRuleExecutionStateStore,
|
||||
automationRunResultToApiBody,
|
||||
getSingleAutomationEmailSendResult,
|
||||
runAutomationRuleForManualRoute,
|
||||
runAutomationRuleForRoute,
|
||||
} from "@/lib/automations/run-route";
|
||||
import { parseAutomationScheduledAtMillis } from "@/lib/automations/scheduled-at";
|
||||
@ -37,15 +40,29 @@ describe("automation scheduled timestamp parsing", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("automation email enqueue result classification", () => {
|
||||
it("distinguishes newly created and already-enqueued single-recipient rows", () => {
|
||||
expect(getSingleAutomationEmailSendResult({ createdCount: 1, alreadyEnqueuedCount: 0 })).toEqual({ outcome: "created" });
|
||||
expect(getSingleAutomationEmailSendResult({ createdCount: 0, alreadyEnqueuedCount: 1 })).toEqual({ outcome: "already-enqueued" });
|
||||
});
|
||||
|
||||
it("fails loudly for partial or otherwise invalid single-recipient results", () => {
|
||||
expect(() => getSingleAutomationEmailSendResult({ createdCount: 1, alreadyEnqueuedCount: 1 }))
|
||||
.toThrow(/Expected one automation email enqueue result/);
|
||||
});
|
||||
});
|
||||
|
||||
type InMemoryExecutionState = {
|
||||
lastTriggeredAt: Date,
|
||||
lastActionAt: Date | null,
|
||||
lastEmailOutboxId: string | null,
|
||||
emailOutboxId: string,
|
||||
lastSourceSnapshot: Record<string, unknown>,
|
||||
};
|
||||
|
||||
function createInMemoryStateStore() {
|
||||
const states = new Map<string, InMemoryExecutionState>();
|
||||
let nextEmailOutboxId = 1;
|
||||
const createEmailOutboxId = () => `00000000-0000-4000-8000-${String(nextEmailOutboxId++).padStart(12, "0")}`;
|
||||
const keyFor = (options: {
|
||||
tenancyId: string,
|
||||
ruleId: string,
|
||||
@ -55,7 +72,7 @@ function createInMemoryStateStore() {
|
||||
}) => `${options.tenancyId}:${options.ruleId}:${options.subjectType}:${options.subjectId}:${options.signalKey}`;
|
||||
|
||||
const stateStore: AutomationRuleExecutionStateStore = {
|
||||
claimExecution: vi.fn(async (options) => {
|
||||
claimExecution: vi.fn(async (options): Promise<Awaited<ReturnType<AutomationRuleExecutionStateStore["claimExecution"]>>> => {
|
||||
const key = keyFor(options);
|
||||
const existing = states.get(key);
|
||||
const cooldownCutoff = new Date(options.lastTriggeredAt.getTime() - options.cooldownDays * 24 * 60 * 60 * 1000);
|
||||
@ -73,15 +90,17 @@ function createInMemoryStateStore() {
|
||||
};
|
||||
}
|
||||
|
||||
const emailOutboxId = existing?.lastActionAt === null ? existing.emailOutboxId : createEmailOutboxId();
|
||||
states.set(key, {
|
||||
lastTriggeredAt: options.lastTriggeredAt,
|
||||
lastActionAt: null,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId,
|
||||
lastSourceSnapshot: options.sourceSnapshot,
|
||||
});
|
||||
return {
|
||||
claimed: true,
|
||||
lastActionAt: existing?.lastActionAt ?? null,
|
||||
emailOutboxId,
|
||||
};
|
||||
}),
|
||||
markActionCompleted: vi.fn(async (options) => {
|
||||
@ -93,7 +112,7 @@ function createInMemoryStateStore() {
|
||||
states.set(key, {
|
||||
...existing,
|
||||
lastActionAt: options.lastActionAt,
|
||||
lastEmailOutboxId: options.lastEmailOutboxId,
|
||||
emailOutboxId: options.emailOutboxId,
|
||||
});
|
||||
}),
|
||||
};
|
||||
@ -115,7 +134,7 @@ async function runWithFakes(options: {
|
||||
} = {}) {
|
||||
const sourceAdapter = createAutomationRouteTestSourceAdapter(options.decisionFactory, { evaluatedCount: 1 });
|
||||
const actionAdapter = createAutomationRouteTestActionAdapter();
|
||||
const emailSender = vi.fn(options.emailSender ?? (async () => {}));
|
||||
const emailSender = vi.fn(options.emailSender ?? (async () => ({ outcome: "created" as const })));
|
||||
let createdStore: ReturnType<typeof createInMemoryStateStore> | undefined;
|
||||
let stateStore = options.stateStore;
|
||||
if (stateStore === undefined) {
|
||||
@ -149,10 +168,10 @@ describe("automation real-send route helpers", () => {
|
||||
it("returns 404 for missing manual rules before evaluating, claiming state, or sending email", async () => {
|
||||
const sourceAdapter = createAutomationRouteTestSourceAdapter();
|
||||
const actionAdapter = createAutomationRouteTestActionAdapter();
|
||||
const emailSender = vi.fn(async () => {});
|
||||
const emailSender = vi.fn(async () => ({ outcome: "created" as const }));
|
||||
const { stateStore } = createInMemoryStateStore();
|
||||
|
||||
const resultPromise = runAutomationRuleForRoute({
|
||||
const resultPromise = runAutomationRuleForManualRoute({
|
||||
tenancy: createAutomationRouteTestTenancy({ ruleExists: false }),
|
||||
ruleId,
|
||||
scheduledAt,
|
||||
@ -175,10 +194,10 @@ describe("automation real-send route helpers", () => {
|
||||
it("refuses disabled rules before evaluating, claiming state, or sending email", async () => {
|
||||
const sourceAdapter = createAutomationRouteTestSourceAdapter();
|
||||
const actionAdapter = createAutomationRouteTestActionAdapter();
|
||||
const emailSender = vi.fn(async () => {});
|
||||
const emailSender = vi.fn(async () => ({ outcome: "created" as const }));
|
||||
const { stateStore } = createInMemoryStateStore();
|
||||
|
||||
await expect(runAutomationRuleForRoute({
|
||||
await expect(runAutomationRuleForManualRoute({
|
||||
tenancy: createAutomationRouteTestTenancy({ enabled: false }),
|
||||
ruleId,
|
||||
scheduledAt,
|
||||
@ -215,7 +234,7 @@ describe("automation real-send route helpers", () => {
|
||||
}));
|
||||
expect([...states!.values()]).toMatchObject([{
|
||||
lastActionAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: expect.any(String),
|
||||
lastSourceSnapshot: expect.objectContaining({
|
||||
itemId: "api_credits",
|
||||
thresholdKind: "near",
|
||||
@ -254,7 +273,7 @@ describe("automation real-send route helpers", () => {
|
||||
states.set("tenancy-1:low-api-credits:user:user-1:api_credits:near", {
|
||||
lastTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: null,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: "00000000-0000-4000-8000-000000000001",
|
||||
lastSourceSnapshot: createAutomationRouteTestSourceDecision().sourceSnapshot,
|
||||
});
|
||||
|
||||
@ -346,7 +365,7 @@ describe("automation real-send route helpers", () => {
|
||||
expect([...states.values()]).toMatchObject([{
|
||||
lastTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: null,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: expect.any(String),
|
||||
}]);
|
||||
|
||||
const immediateRetry = await runWithFakes({
|
||||
@ -378,6 +397,54 @@ describe("automation real-send route helpers", () => {
|
||||
expect(retryAfterStaleClaim.emailSender).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("does not enqueue a duplicate when completion fails after the outbox row was created", async () => {
|
||||
const { prisma, rows } = createMockExecutionStatePrisma();
|
||||
const reservedEmailOutboxId = "00000000-0000-4000-8000-000000000001";
|
||||
const prismaStore = createPrismaAutomationRuleExecutionStateStore(prisma, {
|
||||
createEmailOutboxId: () => reservedEmailOutboxId,
|
||||
});
|
||||
let shouldFailCompletion = true;
|
||||
const stateStore: AutomationRuleExecutionStateStore = {
|
||||
claimExecution: prismaStore.claimExecution,
|
||||
markActionCompleted: vi.fn(async (options) => {
|
||||
if (shouldFailCompletion) {
|
||||
shouldFailCompletion = false;
|
||||
throw new Error("completion database write failed");
|
||||
}
|
||||
await prismaStore.markActionCompleted(options);
|
||||
}),
|
||||
};
|
||||
const enqueuedEmailOutboxIds = new Set<string>();
|
||||
const emailSender = vi.fn(async (options: Parameters<AutomationEmailSender>[0]) => {
|
||||
if (enqueuedEmailOutboxIds.has(options.emailOutboxId)) {
|
||||
return { outcome: "already-enqueued" as const };
|
||||
}
|
||||
enqueuedEmailOutboxIds.add(options.emailOutboxId);
|
||||
return { outcome: "created" as const };
|
||||
});
|
||||
|
||||
await expect(runWithFakes({
|
||||
stateStore,
|
||||
emailSender,
|
||||
})).rejects.toThrow("completion database write failed");
|
||||
|
||||
const retry = await runWithFakes({
|
||||
stateStore,
|
||||
emailSender,
|
||||
now: new Date("2026-07-01T12:16:00.000Z"),
|
||||
});
|
||||
|
||||
expect(retry.result.sentCount).toBe(1);
|
||||
expect(emailSender).toHaveBeenCalledTimes(2);
|
||||
expect(emailSender).toHaveBeenNthCalledWith(1, expect.objectContaining({ emailOutboxId: reservedEmailOutboxId }));
|
||||
expect(emailSender).toHaveBeenNthCalledWith(2, expect.objectContaining({ emailOutboxId: reservedEmailOutboxId }));
|
||||
expect(enqueuedEmailOutboxIds).toEqual(new Set([reservedEmailOutboxId]));
|
||||
expect([...rows.values()]).toMatchObject([{
|
||||
emailOutboxId: reservedEmailOutboxId,
|
||||
lastActionAt: new Date("2026-07-01T12:16:00.000Z"),
|
||||
}]);
|
||||
});
|
||||
|
||||
it("suppresses a second run after an expired cooldown is reclaimed but not completed", async () => {
|
||||
const { stateStore } = createInMemoryStateStore();
|
||||
|
||||
@ -411,7 +478,7 @@ type PrismaStoreState = {
|
||||
signalKey: string,
|
||||
lastTriggeredAt: Date,
|
||||
lastActionAt: Date | null,
|
||||
lastEmailOutboxId: string | null,
|
||||
emailOutboxId: string,
|
||||
lastSourceSnapshot: Prisma.InputJsonObject,
|
||||
};
|
||||
|
||||
@ -451,6 +518,7 @@ function createMockExecutionStatePrisma(initialRows: PrismaStoreState[] = []) {
|
||||
return row === undefined ? null : {
|
||||
lastTriggeredAt: row.lastTriggeredAt,
|
||||
lastActionAt: row.lastActionAt,
|
||||
emailOutboxId: row.emailOutboxId,
|
||||
};
|
||||
}),
|
||||
updateMany: vi.fn(async (options) => {
|
||||
@ -459,13 +527,16 @@ function createMockExecutionStatePrisma(initialRows: PrismaStoreState[] = []) {
|
||||
return { count: 0 };
|
||||
}
|
||||
|
||||
const canClaim = options.where.OR.some((clause) => {
|
||||
if ("lastTriggeredAt" in clause) {
|
||||
return row.lastActionAt === null && row.lastTriggeredAt < clause.lastTriggeredAt.lt;
|
||||
}
|
||||
return row.lastActionAt !== null && row.lastActionAt < clause.lastActionAt.lt;
|
||||
});
|
||||
if (!canClaim) {
|
||||
const lastActionMatches = options.where.lastActionAt === null
|
||||
? row.lastActionAt === null
|
||||
: row.lastActionAt !== null && row.lastActionAt < options.where.lastActionAt.lt;
|
||||
const lastTriggeredMatches = options.where.lastTriggeredAt === undefined
|
||||
|| (options.where.lastTriggeredAt instanceof Date
|
||||
? row.lastTriggeredAt.getTime() === options.where.lastTriggeredAt.getTime()
|
||||
: row.lastTriggeredAt < options.where.lastTriggeredAt.lt);
|
||||
const emailOutboxIdMatches = options.where.emailOutboxId === undefined
|
||||
|| row.emailOutboxId === options.where.emailOutboxId;
|
||||
if (!lastActionMatches || !lastTriggeredMatches || !emailOutboxIdMatches) {
|
||||
return { count: 0 };
|
||||
}
|
||||
|
||||
@ -475,18 +546,6 @@ function createMockExecutionStatePrisma(initialRows: PrismaStoreState[] = []) {
|
||||
});
|
||||
return { count: 1 };
|
||||
}),
|
||||
update: vi.fn(async (options) => {
|
||||
const key = keyFor(options.where.tenancyId_ruleId_subjectType_subjectId_signalKey);
|
||||
const row = rows.get(key);
|
||||
if (row === undefined) {
|
||||
throw new Error("Expected automation execution state row to exist before update.");
|
||||
}
|
||||
rows.set(key, {
|
||||
...row,
|
||||
...options.data,
|
||||
});
|
||||
return rows.get(key);
|
||||
}),
|
||||
},
|
||||
};
|
||||
|
||||
@ -516,19 +575,22 @@ function createClaimOptions(options: {
|
||||
describe("Prisma automation execution state store", () => {
|
||||
it("claims a new execution state row with Prisma create", async () => {
|
||||
const { prisma, rows } = createMockExecutionStatePrisma();
|
||||
const store = createPrismaAutomationRuleExecutionStateStore(prisma);
|
||||
const store = createPrismaAutomationRuleExecutionStateStore(prisma, {
|
||||
createEmailOutboxId: () => "00000000-0000-4000-8000-000000000001",
|
||||
});
|
||||
|
||||
const result = await store.claimExecution(createClaimOptions());
|
||||
|
||||
expect(result).toEqual({
|
||||
claimed: true,
|
||||
lastActionAt: null,
|
||||
emailOutboxId: "00000000-0000-4000-8000-000000000001",
|
||||
});
|
||||
expect(prisma.automationRuleExecutionState.create).toHaveBeenCalledOnce();
|
||||
expect([...rows.values()]).toMatchObject([{
|
||||
lastTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: null,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: "00000000-0000-4000-8000-000000000001",
|
||||
lastSourceSnapshot: expect.objectContaining({
|
||||
itemId: "api_credits",
|
||||
}),
|
||||
@ -543,7 +605,7 @@ describe("Prisma automation execution state store", () => {
|
||||
actionType: "send-email",
|
||||
lastTriggeredAt: new Date("2026-06-01T12:00:00.000Z"),
|
||||
lastActionAt: oldActionAt,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: "00000000-0000-4000-8000-000000000001",
|
||||
lastSourceSnapshot: {
|
||||
itemId: "old",
|
||||
},
|
||||
@ -555,6 +617,7 @@ describe("Prisma automation execution state store", () => {
|
||||
expect(result).toEqual({
|
||||
claimed: true,
|
||||
lastActionAt: oldActionAt,
|
||||
emailOutboxId: expect.any(String),
|
||||
});
|
||||
expect(prisma.automationRuleExecutionState.updateMany).toHaveBeenCalledOnce();
|
||||
});
|
||||
@ -567,7 +630,7 @@ describe("Prisma automation execution state store", () => {
|
||||
actionType: "send-email",
|
||||
lastTriggeredAt: new Date("2026-06-30T12:00:00.000Z"),
|
||||
lastActionAt,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: "00000000-0000-4000-8000-000000000001",
|
||||
lastSourceSnapshot: {
|
||||
itemId: "api_credits",
|
||||
},
|
||||
@ -597,6 +660,7 @@ describe("Prisma automation execution state store", () => {
|
||||
expect(first).toEqual({
|
||||
claimed: true,
|
||||
lastActionAt: null,
|
||||
emailOutboxId: expect.any(String),
|
||||
});
|
||||
expect(second).toEqual({
|
||||
claimed: false,
|
||||
@ -608,7 +672,9 @@ describe("Prisma automation execution state store", () => {
|
||||
|
||||
it("continues to apply cooldown and retry rules after a claim completes", async () => {
|
||||
const { prisma } = createMockExecutionStatePrisma();
|
||||
const store = createPrismaAutomationRuleExecutionStateStore(prisma);
|
||||
const store = createPrismaAutomationRuleExecutionStateStore(prisma, {
|
||||
createEmailOutboxId: () => "00000000-0000-4000-8000-000000000001",
|
||||
});
|
||||
|
||||
await store.claimExecution(createClaimOptions({
|
||||
lastTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
@ -619,8 +685,9 @@ describe("Prisma automation execution state store", () => {
|
||||
subjectType: "user",
|
||||
subjectId: "user-1",
|
||||
signalKey: "api_credits:near",
|
||||
claimTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: new Date("2026-07-01T12:05:00.000Z"),
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: "00000000-0000-4000-8000-000000000001",
|
||||
});
|
||||
|
||||
await expect(store.claimExecution(createClaimOptions({
|
||||
@ -634,6 +701,7 @@ describe("Prisma automation execution state store", () => {
|
||||
}))).resolves.toEqual({
|
||||
claimed: true,
|
||||
lastActionAt: new Date("2026-07-01T12:05:00.000Z"),
|
||||
emailOutboxId: expect.any(String),
|
||||
});
|
||||
});
|
||||
|
||||
@ -645,7 +713,7 @@ describe("Prisma automation execution state store", () => {
|
||||
actionType: "send-email",
|
||||
lastTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: "00000000-0000-4000-8000-000000000001",
|
||||
lastSourceSnapshot: {
|
||||
itemId: "api_credits",
|
||||
},
|
||||
@ -662,6 +730,7 @@ describe("Prisma automation execution state store", () => {
|
||||
expect(first).toEqual({
|
||||
claimed: true,
|
||||
lastActionAt,
|
||||
emailOutboxId: expect.any(String),
|
||||
});
|
||||
expect(second).toEqual({
|
||||
claimed: false,
|
||||
@ -673,14 +742,58 @@ describe("Prisma automation execution state store", () => {
|
||||
}]);
|
||||
});
|
||||
|
||||
it("marks a claimed action completed with Prisma update", async () => {
|
||||
it("reuses an in-flight reservation after staleness and fences the previous claimant", async () => {
|
||||
const reservedEmailOutboxId = "00000000-0000-4000-8000-000000000001";
|
||||
const { prisma } = createMockExecutionStatePrisma([{
|
||||
...createClaimOptions(),
|
||||
sourceType: "payments-item-quota",
|
||||
actionType: "send-email",
|
||||
lastTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: null,
|
||||
emailOutboxId: reservedEmailOutboxId,
|
||||
lastSourceSnapshot: { itemId: "api_credits" },
|
||||
}]);
|
||||
const store = createPrismaAutomationRuleExecutionStateStore(prisma);
|
||||
|
||||
await expect(store.claimExecution(createClaimOptions({
|
||||
lastTriggeredAt: new Date("2026-07-01T12:16:00.000Z"),
|
||||
}))).resolves.toEqual({
|
||||
claimed: true,
|
||||
lastActionAt: null,
|
||||
emailOutboxId: reservedEmailOutboxId,
|
||||
});
|
||||
|
||||
await expect(store.markActionCompleted({
|
||||
tenancyId: "tenancy-1",
|
||||
ruleId,
|
||||
subjectType: "user",
|
||||
subjectId: "user-1",
|
||||
signalKey: "api_credits:near",
|
||||
claimTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: new Date("2026-07-01T12:17:00.000Z"),
|
||||
emailOutboxId: reservedEmailOutboxId,
|
||||
})).rejects.toThrow("lost ownership");
|
||||
|
||||
await expect(store.markActionCompleted({
|
||||
tenancyId: "tenancy-1",
|
||||
ruleId,
|
||||
subjectType: "user",
|
||||
subjectId: "user-1",
|
||||
signalKey: "api_credits:near",
|
||||
claimTriggeredAt: new Date("2026-07-01T12:16:00.000Z"),
|
||||
lastActionAt: new Date("2026-07-01T12:16:00.000Z"),
|
||||
emailOutboxId: reservedEmailOutboxId,
|
||||
})).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("marks a claimed action completed with a fenced Prisma updateMany", async () => {
|
||||
const { prisma, rows } = createMockExecutionStatePrisma([{
|
||||
...createClaimOptions(),
|
||||
sourceType: "payments-item-quota",
|
||||
actionType: "send-email",
|
||||
lastTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: null,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: "9ddfd5da-8cca-48be-944a-f59235892877",
|
||||
lastSourceSnapshot: {
|
||||
itemId: "api_credits",
|
||||
},
|
||||
@ -693,14 +806,26 @@ describe("Prisma automation execution state store", () => {
|
||||
subjectType: "user",
|
||||
subjectId: "user-1",
|
||||
signalKey: "api_credits:near",
|
||||
claimTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: new Date("2026-07-01T12:05:00.000Z"),
|
||||
lastEmailOutboxId: "9ddfd5da-8cca-48be-944a-f59235892877",
|
||||
emailOutboxId: "9ddfd5da-8cca-48be-944a-f59235892877",
|
||||
});
|
||||
|
||||
expect(prisma.automationRuleExecutionState.update).toHaveBeenCalledOnce();
|
||||
expect(prisma.automationRuleExecutionState.updateMany).toHaveBeenCalledOnce();
|
||||
expect([...rows.values()]).toMatchObject([{
|
||||
lastActionAt: new Date("2026-07-01T12:05:00.000Z"),
|
||||
lastEmailOutboxId: "9ddfd5da-8cca-48be-944a-f59235892877",
|
||||
emailOutboxId: "9ddfd5da-8cca-48be-944a-f59235892877",
|
||||
}]);
|
||||
|
||||
await expect(store.markActionCompleted({
|
||||
tenancyId: "tenancy-1",
|
||||
ruleId,
|
||||
subjectType: "user",
|
||||
subjectId: "user-1",
|
||||
signalKey: "api_credits:near",
|
||||
claimTriggeredAt: new Date("2026-07-01T12:00:00.000Z"),
|
||||
lastActionAt: new Date("2026-07-01T12:05:00.000Z"),
|
||||
emailOutboxId: "9ddfd5da-8cca-48be-944a-f59235892877",
|
||||
})).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
@ -2,7 +2,8 @@ import { createSendEmailActionAdapter } from "@/lib/automations/actions/send-ema
|
||||
import { createPrismaAutomationRuleExecutionStateStore } from "@/lib/automations/execution-state-store";
|
||||
import {
|
||||
automationRunResultToApiBody,
|
||||
runAutomationRuleForRoute,
|
||||
getSingleAutomationEmailSendResult,
|
||||
runAutomationRuleForManualRoute,
|
||||
} from "@/lib/automations/run-route";
|
||||
import { parseAutomationScheduledAtMillis } from "@/lib/automations/scheduled-at";
|
||||
import {
|
||||
@ -80,7 +81,7 @@ export const POST = createSmartRouteHandler({
|
||||
customerDataReaders: paymentsItemQuotaCustomerDataReaders,
|
||||
});
|
||||
const scheduledAt = parseAutomationScheduledAtMillis(body.scheduled_at_millis, "scheduled_at_millis");
|
||||
const result = await runAutomationRuleForRoute({
|
||||
const result = await runAutomationRuleForManualRoute({
|
||||
tenancy: auth.tenancy,
|
||||
ruleId: params.rule_id,
|
||||
limit: body.limit,
|
||||
@ -90,8 +91,8 @@ export const POST = createSmartRouteHandler({
|
||||
sourceAdapter,
|
||||
actionAdapter: createSendEmailActionAdapter(),
|
||||
stateStore: createPrismaAutomationRuleExecutionStateStore(prisma),
|
||||
emailSender: async ({ action, scheduledAt }) => {
|
||||
await sendEmailToMany({
|
||||
emailSender: async ({ action, scheduledAt, emailOutboxId }) => {
|
||||
const enqueueResult = await sendEmailToMany({
|
||||
tenancy: auth.tenancy,
|
||||
recipients: [action.recipient],
|
||||
tsxSource: action.tsxSource,
|
||||
@ -103,7 +104,9 @@ export const POST = createSmartRouteHandler({
|
||||
createdWith: action.createdWith,
|
||||
overrideSubject: action.subject,
|
||||
overrideNotificationCategoryId: action.notificationCategoryId,
|
||||
emailOutboxIds: [emailOutboxId],
|
||||
});
|
||||
return getSingleAutomationEmailSendResult(enqueueResult);
|
||||
},
|
||||
});
|
||||
|
||||
|
||||
@ -1,10 +1,12 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { scheduledAutomationWorkBudgetMs } from "@/lib/automations/scheduler";
|
||||
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);
|
||||
expect(scheduledAutomationWorkBudgetMs).toBeLessThan(maxDuration * 1000);
|
||||
});
|
||||
|
||||
it("allows omitted and lower positive integer bounds", () => {
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
import { AutomationActionAdapter, AutomationActionPlan, AutomationSourceDecision } from "../rule-evaluator";
|
||||
import { AutomationRuleTenancy, PaymentsItemQuotaAutomationRule, sendEmailActionType } from "../rules";
|
||||
import { AutomationRuleTenancy, NonRetryableAutomationRuleError, PaymentsItemQuotaAutomationRule, sendEmailActionType } from "../rules";
|
||||
|
||||
export type SendEmailNotificationCategory = {
|
||||
id: string,
|
||||
@ -35,15 +35,15 @@ export async function buildSendEmailActionPlan(options: {
|
||||
}): Promise<AutomationActionPlan> {
|
||||
const template = options.tenancy.config.emails?.templates?.[options.rule.action.templateId];
|
||||
if (template === undefined) {
|
||||
throw new Error(`Automation rule "${options.ruleId}" references email template "${options.rule.action.templateId}", but that template does not exist.`);
|
||||
throw new NonRetryableAutomationRuleError("missing-template", `Automation rule "${options.ruleId}" references email template "${options.rule.action.templateId}", but that template does not exist.`);
|
||||
}
|
||||
if (template.tsxSource === undefined) {
|
||||
throw new Error(`Automation rule "${options.ruleId}" references email template "${options.rule.action.templateId}", but that template is missing tsxSource.`);
|
||||
throw new NonRetryableAutomationRuleError("invalid-template", `Automation rule "${options.ruleId}" references email template "${options.rule.action.templateId}", but that template is missing tsxSource.`);
|
||||
}
|
||||
|
||||
const notificationCategoryName: string = options.rule.action.notificationCategoryName ?? "Marketing";
|
||||
if (notificationCategoryName !== "Marketing") {
|
||||
throw new Error(`Automation rule "${options.ruleId}" has unsupported action.notificationCategoryName "${notificationCategoryName}". V1 supports only "Marketing".`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${options.ruleId}" has unsupported action.notificationCategoryName "${notificationCategoryName}". V1 supports only "Marketing".`);
|
||||
}
|
||||
const notificationCategoryReader = options.getNotificationCategoryByName ?? getNotificationCategoryByNameFromExistingEmailSystem;
|
||||
const notificationCategory = await notificationCategoryReader(notificationCategoryName);
|
||||
@ -53,7 +53,7 @@ export async function buildSendEmailActionPlan(options: {
|
||||
|
||||
const projectDisplayName = options.tenancy.project?.display_name;
|
||||
if (projectDisplayName === undefined) {
|
||||
throw new Error(`Automation rule "${options.ruleId}" cannot build email variables because tenancy.project.display_name is missing.`);
|
||||
throw new NonRetryableAutomationRuleError("invalid-rule-context", `Automation rule "${options.ruleId}" cannot build email variables because tenancy.project.display_name is missing.`);
|
||||
}
|
||||
|
||||
return {
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { Prisma } from "@/generated/prisma/client";
|
||||
import { getAutomationCooldownStatus, type AutomationCooldownStatus } from "./cooldown";
|
||||
import { AutomationRuleExecutionStateStore } from "./run-route";
|
||||
@ -19,7 +20,7 @@ export type AutomationRuleExecutionStatePrisma = {
|
||||
actionType: string,
|
||||
lastTriggeredAt: Date,
|
||||
lastActionAt: Date | null,
|
||||
lastEmailOutboxId: string | null,
|
||||
emailOutboxId: string,
|
||||
lastSourceSnapshot: Prisma.InputJsonObject,
|
||||
},
|
||||
}) => Promise<unknown>,
|
||||
@ -30,35 +31,24 @@ export type AutomationRuleExecutionStatePrisma = {
|
||||
select: {
|
||||
lastTriggeredAt: true,
|
||||
lastActionAt: true,
|
||||
emailOutboxId: true,
|
||||
},
|
||||
}) => Promise<{ lastTriggeredAt: Date, lastActionAt: Date | null } | null>,
|
||||
}) => Promise<{ lastTriggeredAt: Date, lastActionAt: Date | null, emailOutboxId: string } | null>,
|
||||
updateMany: (options: {
|
||||
where: AutomationRuleExecutionStateKey & {
|
||||
OR: Array<
|
||||
| {
|
||||
lastActionAt: null,
|
||||
lastTriggeredAt: { lt: Date },
|
||||
}
|
||||
| { lastActionAt: { lt: Date } }
|
||||
>,
|
||||
lastActionAt: null | { lt: Date },
|
||||
lastTriggeredAt?: Date | { lt: Date },
|
||||
emailOutboxId?: string,
|
||||
},
|
||||
data: {
|
||||
sourceType: string,
|
||||
actionType: string,
|
||||
lastTriggeredAt: Date,
|
||||
lastActionAt: Date | null,
|
||||
lastSourceSnapshot: Prisma.InputJsonObject,
|
||||
sourceType?: string,
|
||||
actionType?: string,
|
||||
lastTriggeredAt?: Date,
|
||||
lastActionAt?: Date | null,
|
||||
emailOutboxId?: string,
|
||||
lastSourceSnapshot?: Prisma.InputJsonObject,
|
||||
},
|
||||
}) => Promise<{ count: number }>,
|
||||
update: (options: {
|
||||
where: {
|
||||
tenancyId_ruleId_subjectType_subjectId_signalKey: AutomationRuleExecutionStateKey,
|
||||
},
|
||||
data: {
|
||||
lastActionAt: Date,
|
||||
lastEmailOutboxId: string | null,
|
||||
},
|
||||
}) => Promise<unknown>,
|
||||
},
|
||||
};
|
||||
|
||||
@ -103,7 +93,11 @@ export function createPrismaAutomationRuleExecutionStateReader(prisma: Automatio
|
||||
};
|
||||
}
|
||||
|
||||
export function createPrismaAutomationRuleExecutionStateStore(prisma: AutomationRuleExecutionStatePrisma): AutomationRuleExecutionStateStore {
|
||||
export function createPrismaAutomationRuleExecutionStateStore(
|
||||
prisma: AutomationRuleExecutionStatePrisma,
|
||||
options: { createEmailOutboxId?: () => string } = {},
|
||||
): AutomationRuleExecutionStateStore {
|
||||
const createEmailOutboxId = options.createEmailOutboxId ?? randomUUID;
|
||||
return {
|
||||
async claimExecution(options) {
|
||||
const cooldownCutoff = new Date(options.lastTriggeredAt.getTime() - options.cooldownDays * 24 * 60 * 60 * 1000);
|
||||
@ -112,6 +106,7 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
const staleClaimCutoff = new Date(options.lastTriggeredAt.getTime() - 15 * 60 * 1000);
|
||||
const stateKey = getAutomationRuleExecutionStateKey(options);
|
||||
const lastSourceSnapshot = getPrismaJsonObject(options.sourceSnapshot);
|
||||
const initialEmailOutboxId = createEmailOutboxId();
|
||||
|
||||
try {
|
||||
await prisma.automationRuleExecutionState.create({
|
||||
@ -125,7 +120,7 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
signalKey: options.signalKey,
|
||||
lastTriggeredAt: options.lastTriggeredAt,
|
||||
lastActionAt: null,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: initialEmailOutboxId,
|
||||
lastSourceSnapshot,
|
||||
},
|
||||
});
|
||||
@ -133,6 +128,7 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
return {
|
||||
claimed: true,
|
||||
lastActionAt: null,
|
||||
emailOutboxId: initialEmailOutboxId,
|
||||
};
|
||||
} catch (error) {
|
||||
if (!isPrismaUniqueConstraintError(error)) {
|
||||
@ -147,12 +143,15 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
select: {
|
||||
lastTriggeredAt: true,
|
||||
lastActionAt: true,
|
||||
emailOutboxId: true,
|
||||
},
|
||||
});
|
||||
if (existing === null) {
|
||||
throw new Error("Automation rule execution state disappeared after a unique constraint conflict.");
|
||||
}
|
||||
|
||||
// Stale in-flight retries must address the same outbox row; a completed cooldown cycle needs a new row.
|
||||
const emailOutboxId = existing.lastActionAt === null ? existing.emailOutboxId : createEmailOutboxId();
|
||||
const updateResult = await prisma.automationRuleExecutionState.updateMany({
|
||||
where: {
|
||||
tenancyId: options.tenancyId,
|
||||
@ -160,25 +159,15 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
subjectType: options.subjectType,
|
||||
subjectId: options.subjectId,
|
||||
signalKey: options.signalKey,
|
||||
OR: [
|
||||
{
|
||||
lastActionAt: null,
|
||||
lastTriggeredAt: {
|
||||
lt: staleClaimCutoff,
|
||||
},
|
||||
},
|
||||
{
|
||||
lastActionAt: {
|
||||
lt: cooldownCutoff,
|
||||
},
|
||||
},
|
||||
],
|
||||
lastActionAt: existing.lastActionAt === null ? null : { lt: cooldownCutoff },
|
||||
...(existing.lastActionAt === null ? { lastTriggeredAt: { lt: staleClaimCutoff } } : {}),
|
||||
},
|
||||
data: {
|
||||
sourceType: options.sourceType,
|
||||
actionType: options.actionType,
|
||||
lastTriggeredAt: options.lastTriggeredAt,
|
||||
lastActionAt: null,
|
||||
emailOutboxId,
|
||||
lastSourceSnapshot,
|
||||
},
|
||||
});
|
||||
@ -186,6 +175,7 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
return {
|
||||
claimed: true,
|
||||
lastActionAt: existing.lastActionAt,
|
||||
emailOutboxId,
|
||||
};
|
||||
}
|
||||
if (updateResult.count !== 0) {
|
||||
@ -199,6 +189,7 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
select: {
|
||||
lastTriggeredAt: true,
|
||||
lastActionAt: true,
|
||||
emailOutboxId: true,
|
||||
},
|
||||
});
|
||||
if (current === null) {
|
||||
@ -211,15 +202,43 @@ export function createPrismaAutomationRuleExecutionStateStore(prisma: Automation
|
||||
};
|
||||
},
|
||||
async markActionCompleted(options) {
|
||||
await prisma.automationRuleExecutionState.update({
|
||||
const stateKey = getAutomationRuleExecutionStateKey(options);
|
||||
const updateResult = await prisma.automationRuleExecutionState.updateMany({
|
||||
where: {
|
||||
tenancyId_ruleId_subjectType_subjectId_signalKey: getAutomationRuleExecutionStateKey(options),
|
||||
...stateKey,
|
||||
lastActionAt: null,
|
||||
lastTriggeredAt: options.claimTriggeredAt,
|
||||
emailOutboxId: options.emailOutboxId,
|
||||
},
|
||||
data: {
|
||||
lastActionAt: options.lastActionAt,
|
||||
lastEmailOutboxId: options.lastEmailOutboxId,
|
||||
},
|
||||
});
|
||||
if (updateResult.count === 1) {
|
||||
return;
|
||||
}
|
||||
if (updateResult.count !== 0) {
|
||||
throw new Error(`Expected at most one automation execution state row to complete, received ${updateResult.count}.`);
|
||||
}
|
||||
|
||||
const current = await prisma.automationRuleExecutionState.findUnique({
|
||||
where: {
|
||||
tenancyId_ruleId_subjectType_subjectId_signalKey: stateKey,
|
||||
},
|
||||
select: {
|
||||
lastTriggeredAt: true,
|
||||
lastActionAt: true,
|
||||
emailOutboxId: true,
|
||||
},
|
||||
});
|
||||
if (
|
||||
current?.emailOutboxId === options.emailOutboxId
|
||||
&& current.lastTriggeredAt.getTime() === options.claimTriggeredAt.getTime()
|
||||
&& current.lastActionAt !== null
|
||||
) {
|
||||
return;
|
||||
}
|
||||
throw new Error("Automation execution completion lost ownership of its reserved EmailOutbox row.");
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@ -421,11 +421,13 @@ describe("payments item quota source adapter", () => {
|
||||
const tenancy = createSourceTenancy({ itemExists: false });
|
||||
const { adapter } = createSourceAdapterFixture({ projectUserIds: ["user-1"] });
|
||||
|
||||
await expect(adapter.evaluate({
|
||||
const evaluationPromise = adapter.evaluate({
|
||||
tenancy,
|
||||
ruleId,
|
||||
rule: getSupportedAutomationRule(tenancy, ruleId),
|
||||
})).rejects.toThrow('Automation rule "low-api-credits" references payments item "api_credits", but that item does not exist.');
|
||||
});
|
||||
await expect(evaluationPromise).rejects.toMatchObject({ reason: "missing-item" });
|
||||
await expect(evaluationPromise).rejects.toThrow('Automation rule "low-api-credits" references payments item "api_credits", but that item does not exist.');
|
||||
});
|
||||
|
||||
it("rejects a payments item with a non-user customer type", async () => {
|
||||
@ -844,11 +846,13 @@ describe("send-email action adapter", () => {
|
||||
templateId: "1b477fe1-7479-4d90-ac47-23d0a5048bc8",
|
||||
}));
|
||||
|
||||
await expect(buildTestSendEmailActionPlan({
|
||||
const planPromise = buildTestSendEmailActionPlan({
|
||||
tenancy,
|
||||
ruleId,
|
||||
rule: getSupportedAutomationRule(tenancy, ruleId),
|
||||
decision: createActionDecision(),
|
||||
})).rejects.toThrow('Automation rule "low-api-credits" references email template "1b477fe1-7479-4d90-ac47-23d0a5048bc8", but that template does not exist.');
|
||||
});
|
||||
await expect(planPromise).rejects.toMatchObject({ reason: "missing-template" });
|
||||
await expect(planPromise).rejects.toThrow('Automation rule "low-api-credits" references email template "1b477fe1-7479-4d90-ac47-23d0a5048bc8", but that template does not exist.');
|
||||
});
|
||||
});
|
||||
|
||||
@ -80,6 +80,27 @@ export type AutomationRuleTenancy = {
|
||||
config: AutomationRulesConfig,
|
||||
};
|
||||
|
||||
export type NonRetryableAutomationRuleErrorReason =
|
||||
| "rule-not-found"
|
||||
| "rule-disabled"
|
||||
| "unsupported-rule"
|
||||
| "missing-item"
|
||||
| "incompatible-item"
|
||||
| "missing-template"
|
||||
| "invalid-template"
|
||||
| "invalid-rule-context";
|
||||
|
||||
/** A deterministic rule/configuration failure that will not recover by retrying the same scheduler page. */
|
||||
export class NonRetryableAutomationRuleError extends Error {
|
||||
constructor(
|
||||
readonly reason: NonRetryableAutomationRuleErrorReason,
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = "NonRetryableAutomationRuleError";
|
||||
}
|
||||
}
|
||||
|
||||
export function listAutomationRules(tenancy: AutomationRuleTenancy) {
|
||||
return Object.entries(tenancy.config.automations?.rules ?? {})
|
||||
.filter((entry): entry is [string, AutomationRuleConfig] => entry[1] !== undefined)
|
||||
@ -90,13 +111,20 @@ export function getAutomationRule(tenancy: AutomationRuleTenancy, ruleId: string
|
||||
return tenancy.config.automations?.rules?.[ruleId];
|
||||
}
|
||||
|
||||
export class AutomationRuleNotFoundError extends Error {
|
||||
export class AutomationRuleNotFoundError extends NonRetryableAutomationRuleError {
|
||||
constructor(tenancyId: string, ruleId: string) {
|
||||
super(`Automation rule "${ruleId}" was not found for tenancy "${tenancyId}".`);
|
||||
super("rule-not-found", `Automation rule "${ruleId}" was not found for tenancy "${tenancyId}".`);
|
||||
this.name = "AutomationRuleNotFoundError";
|
||||
}
|
||||
}
|
||||
|
||||
export class AutomationRuleDisabledError extends NonRetryableAutomationRuleError {
|
||||
constructor(ruleId: string) {
|
||||
super("rule-disabled", `Automation rule "${ruleId}" is disabled and cannot be manually sent.`);
|
||||
this.name = "AutomationRuleDisabledError";
|
||||
}
|
||||
}
|
||||
|
||||
export function getSupportedAutomationRule(tenancy: AutomationRuleTenancy, ruleId: string) {
|
||||
const rule = getAutomationRule(tenancy, ruleId);
|
||||
if (rule === undefined) {
|
||||
@ -108,31 +136,31 @@ export function getSupportedAutomationRule(tenancy: AutomationRuleTenancy, ruleI
|
||||
|
||||
export function assertSupportedAutomationRule(ruleId: string, rule: AutomationRuleConfig): asserts rule is SupportedAutomationRule {
|
||||
if (rule.source.type !== paymentsItemQuotaSourceType) {
|
||||
throw new Error(`Automation rule "${ruleId}" has unsupported source.type "${rule.source.type}". V1 supports only "${paymentsItemQuotaSourceType}".`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" has unsupported source.type "${rule.source.type}". V1 supports only "${paymentsItemQuotaSourceType}".`);
|
||||
}
|
||||
if (rule.source.customerType !== userCustomerType) {
|
||||
throw new Error(`Automation rule "${ruleId}" has unsupported source.customerType "${rule.source.customerType ?? "<missing>"}". V1 supports only "${userCustomerType}".`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" has unsupported source.customerType "${rule.source.customerType ?? "<missing>"}". V1 supports only "${userCustomerType}".`);
|
||||
}
|
||||
if (rule.action.type !== sendEmailActionType) {
|
||||
throw new Error(`Automation rule "${ruleId}" has unsupported action.type "${rule.action.type}". V1 supports only "${sendEmailActionType}".`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" has unsupported action.type "${rule.action.type}". V1 supports only "${sendEmailActionType}".`);
|
||||
}
|
||||
if (rule.source.itemId === undefined) {
|
||||
throw new Error(`Automation rule "${ruleId}" is missing source.itemId.`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" is missing source.itemId.`);
|
||||
}
|
||||
if (rule.source.thresholds === undefined) {
|
||||
throw new Error(`Automation rule "${ruleId}" is missing source.thresholds.`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" is missing source.thresholds.`);
|
||||
}
|
||||
if (
|
||||
rule.source.thresholds.nearRemainingRatio === undefined
|
||||
&& rule.source.thresholds.nearRemainingQuantity === undefined
|
||||
&& rule.source.thresholds.overLimitQuantity === undefined
|
||||
) {
|
||||
throw new Error(`Automation rule "${ruleId}" must configure at least one source.thresholds value.`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" must configure at least one source.thresholds value.`);
|
||||
}
|
||||
if (rule.action.templateId === undefined) {
|
||||
throw new Error(`Automation rule "${ruleId}" is missing action.templateId.`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" is missing action.templateId.`);
|
||||
}
|
||||
if (rule.cooldown.days === undefined) {
|
||||
throw new Error(`Automation rule "${ruleId}" is missing cooldown.days.`);
|
||||
throw new NonRetryableAutomationRuleError("unsupported-rule", `Automation rule "${ruleId}" is missing cooldown.days.`);
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
import { StatusError } from "@hexclave/shared/dist/utils/errors";
|
||||
import { AutomationJson, AutomationRuleNotFoundError, AutomationRuleTenancy, getSupportedAutomationRule, paymentsItemQuotaSourceType, sendEmailActionType } from "./rules";
|
||||
import { AutomationJson, AutomationRuleDisabledError, AutomationRuleNotFoundError, getSupportedAutomationRule, paymentsItemQuotaSourceType, sendEmailActionType } from "./rules";
|
||||
import { AutomationActionPlan, AutomationEvaluationResult, AutomationSourceAdapter, AutomationActionAdapter, EvaluatedAutomationDecision, evaluateAutomationRule } from "./rule-evaluator";
|
||||
import { paymentsItemQuotaSourceSnapshotToApiBody } from "./source-snapshot";
|
||||
|
||||
@ -7,6 +7,7 @@ export type AutomationRuleExecutionClaimResult =
|
||||
| {
|
||||
claimed: true,
|
||||
lastActionAt: Date | null,
|
||||
emailOutboxId: string,
|
||||
}
|
||||
| {
|
||||
claimed: false,
|
||||
@ -32,15 +33,34 @@ export type AutomationRuleExecutionStateStore = {
|
||||
subjectType: "user",
|
||||
subjectId: string,
|
||||
signalKey: string,
|
||||
claimTriggeredAt: Date,
|
||||
lastActionAt: Date,
|
||||
lastEmailOutboxId: string | null,
|
||||
emailOutboxId: string,
|
||||
}) => Promise<void>,
|
||||
};
|
||||
|
||||
export type AutomationEmailSendResult = {
|
||||
outcome: "created" | "already-enqueued",
|
||||
};
|
||||
|
||||
export type AutomationEmailSender = (options: {
|
||||
action: AutomationActionPlan,
|
||||
scheduledAt: Date,
|
||||
}) => Promise<void>;
|
||||
emailOutboxId: string,
|
||||
}) => Promise<AutomationEmailSendResult>;
|
||||
|
||||
export function getSingleAutomationEmailSendResult(result: {
|
||||
createdCount: number,
|
||||
alreadyEnqueuedCount: number,
|
||||
}): AutomationEmailSendResult {
|
||||
if (result.createdCount === 1 && result.alreadyEnqueuedCount === 0) {
|
||||
return { outcome: "created" };
|
||||
}
|
||||
if (result.createdCount === 0 && result.alreadyEnqueuedCount === 1) {
|
||||
return { outcome: "already-enqueued" };
|
||||
}
|
||||
throw new Error(`Expected one automation email enqueue result, received ${result.createdCount} created and ${result.alreadyEnqueuedCount} already enqueued.`);
|
||||
}
|
||||
|
||||
export type AutomationRunDecisionResult = {
|
||||
decision: EvaluatedAutomationDecision,
|
||||
@ -76,9 +96,9 @@ export async function runAutomationRuleForRoute(options: {
|
||||
stateStore: AutomationRuleExecutionStateStore,
|
||||
emailSender: AutomationEmailSender,
|
||||
}): Promise<AutomationRunResult> {
|
||||
const rule = getSupportedAutomationRuleForRunRoute(options.tenancy, options.ruleId);
|
||||
const rule = getSupportedAutomationRule(options.tenancy, options.ruleId);
|
||||
if (!rule.enabled) {
|
||||
throw new StatusError(StatusError.Conflict, `Automation rule "${options.ruleId}" is disabled and cannot be manually sent.`);
|
||||
throw new AutomationRuleDisabledError(options.ruleId);
|
||||
}
|
||||
const evaluation = await evaluateAutomationRule({
|
||||
tenancy: options.tenancy,
|
||||
@ -124,6 +144,7 @@ export async function runAutomationRuleForRoute(options: {
|
||||
await options.emailSender({
|
||||
action: decision.action,
|
||||
scheduledAt: options.scheduledAt,
|
||||
emailOutboxId: claim.emailOutboxId,
|
||||
});
|
||||
await options.stateStore.markActionCompleted({
|
||||
tenancyId: options.tenancy.id,
|
||||
@ -131,8 +152,9 @@ export async function runAutomationRuleForRoute(options: {
|
||||
subjectType: decision.subject.type,
|
||||
subjectId: decision.subject.id,
|
||||
signalKey: decision.signal.key,
|
||||
claimTriggeredAt: options.now,
|
||||
lastActionAt: options.now,
|
||||
lastEmailOutboxId: null,
|
||||
emailOutboxId: claim.emailOutboxId,
|
||||
});
|
||||
|
||||
decisions.push({
|
||||
@ -159,13 +181,18 @@ export async function runAutomationRuleForRoute(options: {
|
||||
};
|
||||
}
|
||||
|
||||
function getSupportedAutomationRuleForRunRoute(tenancy: AutomationRuleTenancy, ruleId: string) {
|
||||
export async function runAutomationRuleForManualRoute(
|
||||
options: Parameters<typeof runAutomationRuleForRoute>[0],
|
||||
): Promise<AutomationRunResult> {
|
||||
try {
|
||||
return getSupportedAutomationRule(tenancy, ruleId);
|
||||
return await runAutomationRuleForRoute(options);
|
||||
} catch (error) {
|
||||
if (error instanceof AutomationRuleNotFoundError) {
|
||||
throw new StatusError(StatusError.NotFound, error.message);
|
||||
}
|
||||
if (error instanceof AutomationRuleDisabledError) {
|
||||
throw new StatusError(StatusError.Conflict, error.message);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
@ -14,7 +14,9 @@ vi.mock("@/lib/automations/sources/payments-item-quota", () => ({
|
||||
paymentsItemQuotaCustomerDataReaders: {},
|
||||
prismaPaymentsItemQuotaProjectUserReader: {},
|
||||
}));
|
||||
vi.mock("@/lib/emails", () => ({ sendEmailToMany: async () => {} }));
|
||||
vi.mock("@/lib/emails", () => ({
|
||||
sendEmailToMany: async () => ({ createdCount: 1, alreadyEnqueuedCount: 0 }),
|
||||
}));
|
||||
vi.mock("@/lib/tenancies", () => ({ getTenancy: async () => null }));
|
||||
vi.mock("@/prisma-client", () => ({
|
||||
getPrismaClientForTenancy: async () => ({}),
|
||||
@ -36,6 +38,7 @@ import {
|
||||
runScheduledAutomations,
|
||||
} from "./scheduler";
|
||||
import type { AutomationRunResult } from "./run-route";
|
||||
import { NonRetryableAutomationRuleError } from "./rules";
|
||||
|
||||
const ruleId = "low-api-credits";
|
||||
|
||||
@ -354,6 +357,39 @@ describe("native cron automation traversal", () => {
|
||||
expect(runRule).toHaveBeenCalledWith(expect.objectContaining({ ruleId: validRuleId }));
|
||||
});
|
||||
|
||||
it("advances a deterministic runtime rule failure and runs the next valid rule", async () => {
|
||||
captureErrorMock.mockClear();
|
||||
const state = createStateHarness();
|
||||
const validRuleId = "valid-rule";
|
||||
const runRule = vi.fn(async (options: { ruleId: string }) => {
|
||||
if (options.ruleId === "bad-rule") {
|
||||
throw new NonRetryableAutomationRuleError("missing-template", "Configured template no longer exists.");
|
||||
}
|
||||
return createRunResult(null, { ruleId: options.ruleId });
|
||||
});
|
||||
|
||||
await runScheduledAutomations({
|
||||
prisma: createPrisma(["00000000-0000-4000-8000-000000000010"]),
|
||||
stateStore: state.stateStore,
|
||||
getTenancyById: async (id) => createTenancy(id, {
|
||||
"bad-rule": createRule(),
|
||||
[validRuleId]: createRule(),
|
||||
}),
|
||||
runRule,
|
||||
maxPages: 1,
|
||||
});
|
||||
|
||||
expect(captureErrorMock).toHaveBeenCalledWith("automation-scheduler-non-retryable-rule", expect.any(Error));
|
||||
expect(runRule).toHaveBeenNthCalledWith(1, expect.objectContaining({ ruleId: "bad-rule" }));
|
||||
expect(runRule).toHaveBeenNthCalledWith(2, expect.objectContaining({ ruleId: validRuleId }));
|
||||
expect(state.getCheckpoint()).toMatchObject({
|
||||
activeTenancyId: "00000000-0000-4000-8000-000000000010",
|
||||
completedRuleCursor: validRuleId,
|
||||
activeRuleId: null,
|
||||
nextSubjectCursor: null,
|
||||
});
|
||||
});
|
||||
|
||||
it("skips disabled rules without executing them", async () => {
|
||||
const state = createStateHarness();
|
||||
const runRule = vi.fn(async () => createRunResult(null));
|
||||
|
||||
@ -15,8 +15,8 @@ import { getTenancy, type Tenancy } from "@/lib/tenancies";
|
||||
import { getPrismaClientForTenancy, globalPrismaClient } from "@/prisma-client";
|
||||
import { captureError, HexclaveAssertionError } from "@hexclave/shared/dist/utils/errors";
|
||||
import { stringCompare } from "@hexclave/shared/dist/utils/strings";
|
||||
import { runAutomationRuleForRoute, type AutomationRunResult } from "./run-route";
|
||||
import { assertSupportedAutomationRule, type AutomationRuleTenancy, listAutomationRules } from "./rules";
|
||||
import { getSingleAutomationEmailSendResult, runAutomationRuleForRoute, type AutomationRunResult } from "./run-route";
|
||||
import { assertSupportedAutomationRule, type AutomationRuleTenancy, listAutomationRules, NonRetryableAutomationRuleError } from "./rules";
|
||||
|
||||
export const scheduledAutomationDiscoveryLimit = 500;
|
||||
export const scheduledAutomationRunPageLimit = 100;
|
||||
@ -281,10 +281,14 @@ async function runWithLease<TTenancy extends AutomationRuleTenancy>(options: {
|
||||
try {
|
||||
assertSupportedAutomationRule(activeRuleId, ruleEntry.rule);
|
||||
} catch (error) {
|
||||
if (!(error instanceof NonRetryableAutomationRuleError)) {
|
||||
throw error;
|
||||
}
|
||||
captureError("automation-scheduler-invalid-rule", new HexclaveAssertionError(`Skipping invalid scheduled automation rule "${activeRuleId}" for tenancy "${tenancy.id}".`, {
|
||||
cause: error,
|
||||
tenancyId: tenancy.id,
|
||||
ruleId: activeRuleId,
|
||||
reason: error.reason,
|
||||
}));
|
||||
await saveCheckpoint(completeActiveRule(checkpoint, activeRuleId));
|
||||
rulesProcessed++;
|
||||
@ -299,14 +303,30 @@ async function runWithLease<TTenancy extends AutomationRuleTenancy>(options: {
|
||||
}
|
||||
await options.lease.renewIfNeeded();
|
||||
|
||||
const result = await options.runRule({
|
||||
tenancy,
|
||||
ruleId: activeRuleId,
|
||||
cursor: checkpoint.nextSubjectCursor,
|
||||
limit: options.pageLimit,
|
||||
scheduledAt,
|
||||
now: options.now(),
|
||||
});
|
||||
let result: AutomationRunResult;
|
||||
try {
|
||||
result = await options.runRule({
|
||||
tenancy,
|
||||
ruleId: activeRuleId,
|
||||
cursor: checkpoint.nextSubjectCursor,
|
||||
limit: options.pageLimit,
|
||||
scheduledAt,
|
||||
now: options.now(),
|
||||
});
|
||||
} catch (error) {
|
||||
if (!(error instanceof NonRetryableAutomationRuleError)) {
|
||||
throw error;
|
||||
}
|
||||
captureError("automation-scheduler-non-retryable-rule", new HexclaveAssertionError(`Skipping non-retryable scheduled automation rule "${activeRuleId}" for tenancy "${tenancy.id}".`, {
|
||||
cause: error,
|
||||
tenancyId: tenancy.id,
|
||||
ruleId: activeRuleId,
|
||||
reason: error.reason,
|
||||
}));
|
||||
await saveCheckpoint(completeActiveRule(checkpoint, activeRuleId));
|
||||
rulesProcessed++;
|
||||
continue;
|
||||
}
|
||||
pagesProcessed++;
|
||||
evaluatedCount += result.evaluatedCount;
|
||||
sentCount += result.sentCount;
|
||||
@ -361,8 +381,8 @@ async function runProductionScheduledAutomationRulePage(options: {
|
||||
sourceAdapter,
|
||||
actionAdapter: createSendEmailActionAdapter(),
|
||||
stateStore: createPrismaAutomationRuleExecutionStateStore(prisma),
|
||||
emailSender: async ({ action, scheduledAt }) => {
|
||||
await sendEmailToMany({
|
||||
emailSender: async ({ action, scheduledAt, emailOutboxId }) => {
|
||||
const enqueueResult = await sendEmailToMany({
|
||||
tenancy,
|
||||
recipients: [action.recipient],
|
||||
tsxSource: action.tsxSource,
|
||||
@ -374,7 +394,9 @@ async function runProductionScheduledAutomationRulePage(options: {
|
||||
createdWith: action.createdWith,
|
||||
overrideSubject: action.subject,
|
||||
overrideNotificationCategoryId: action.notificationCategoryId,
|
||||
emailOutboxIds: [emailOutboxId],
|
||||
});
|
||||
return getSingleAutomationEmailSendResult(enqueueResult);
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
@ -5,7 +5,7 @@ import {
|
||||
getSubscriptionMapForCustomer,
|
||||
} from "@/lib/payments/customer-data";
|
||||
import { AutomationSourceAdapter, AutomationSourceDecision, AutomationSourceEvaluationResult } from "../rule-evaluator";
|
||||
import { AutomationRuleTenancy, PaymentsItemQuotaAutomationRule, paymentsItemQuotaSourceType } from "../rules";
|
||||
import { AutomationRuleTenancy, NonRetryableAutomationRuleError, PaymentsItemQuotaAutomationRule, paymentsItemQuotaSourceType } from "../rules";
|
||||
|
||||
export type PaymentsItemQuotaProjectUserPage = {
|
||||
projectUserIds: string[],
|
||||
@ -138,10 +138,10 @@ async function evaluatePaymentsItemQuotaSource<TPrisma>(options: {
|
||||
const itemId = options.rule.source.itemId;
|
||||
const item = options.tenancy.config.payments?.items?.[itemId];
|
||||
if (item === undefined) {
|
||||
throw new Error(`Automation rule "${options.ruleId}" references payments item "${itemId}", but that item does not exist.`);
|
||||
throw new NonRetryableAutomationRuleError("missing-item", `Automation rule "${options.ruleId}" references payments item "${itemId}", but that item does not exist.`);
|
||||
}
|
||||
if (item.customerType !== "user") {
|
||||
throw new Error(`Automation rule "${options.ruleId}" references payments item "${itemId}" with customerType "${item.customerType ?? "<missing>"}"; V1 supports only user items.`);
|
||||
throw new NonRetryableAutomationRuleError("incompatible-item", `Automation rule "${options.ruleId}" references payments item "${itemId}" with customerType "${item.customerType ?? "<missing>"}"; V1 supports only user items.`);
|
||||
}
|
||||
|
||||
const limit = normalizeLimit(options.limit);
|
||||
|
||||
@ -4,7 +4,7 @@ import { EmailOutboxCreatedWith } from '@/generated/prisma/client';
|
||||
import { DEFAULT_TEMPLATE_IDS } from '@hexclave/shared/dist/helpers/emails';
|
||||
import { UsersCrud } from '@hexclave/shared/dist/interface/crud/users';
|
||||
import { getEnvBoolean, getEnvVariable } from '@hexclave/shared/dist/utils/env';
|
||||
import { HexclaveAssertionError } from '@hexclave/shared/dist/utils/errors';
|
||||
import { HexclaveAssertionError, throwErr } from '@hexclave/shared/dist/utils/errors';
|
||||
import { Json } from '@hexclave/shared/dist/utils/json';
|
||||
import { runEmailQueueStep, serializeRecipient } from './email-queue-step';
|
||||
import { LowLevelEmailConfig, isSecureEmailPort } from './emails-low-level';
|
||||
@ -23,6 +23,32 @@ export type EmailOutboxRecipient =
|
||||
| { type: "user-custom-emails", userId: string, emails: string[] }
|
||||
| { type: "custom-emails", emails: string[] };
|
||||
|
||||
export type SendEmailToManyResult = {
|
||||
createdCount: number,
|
||||
alreadyEnqueuedCount: number,
|
||||
};
|
||||
|
||||
function addIdempotencyToEmailOutboxRows<T extends object>(rows: T[], emailOutboxIds: undefined): { data: T[] };
|
||||
function addIdempotencyToEmailOutboxRows<T extends object>(rows: T[], emailOutboxIds: string[]): { data: Array<T & { id: string }>, skipDuplicates: true };
|
||||
function addIdempotencyToEmailOutboxRows<T extends object>(rows: T[], emailOutboxIds: string[] | undefined) {
|
||||
if (emailOutboxIds === undefined) {
|
||||
return { data: rows };
|
||||
}
|
||||
if (emailOutboxIds.length !== rows.length) {
|
||||
throw new HexclaveAssertionError("sendEmailToMany requires exactly one emailOutboxId per recipient.");
|
||||
}
|
||||
if (new Set(emailOutboxIds).size !== emailOutboxIds.length) {
|
||||
throw new HexclaveAssertionError("sendEmailToMany requires unique emailOutboxIds.");
|
||||
}
|
||||
return {
|
||||
data: rows.map((row, index) => ({
|
||||
...row,
|
||||
id: emailOutboxIds[index] ?? throwErr(new HexclaveAssertionError("Missing validated emailOutboxId for recipient.")),
|
||||
})),
|
||||
skipDuplicates: true,
|
||||
};
|
||||
}
|
||||
|
||||
function getDefaultEmailTemplate(tenancy: Tenancy, type: keyof typeof DEFAULT_TEMPLATE_IDS) {
|
||||
const templateList = new Map(Object.entries(tenancy.config.emails.templates));
|
||||
const defaultTemplateIdsMap = new Map(Object.entries(DEFAULT_TEMPLATE_IDS));
|
||||
@ -49,30 +75,38 @@ export async function sendEmailToMany(options: {
|
||||
createdWith: { type: "draft", draftId: string } | { type: "programmatic-call", templateId: string | null },
|
||||
overrideSubject?: string,
|
||||
overrideNotificationCategoryId?: string,
|
||||
}) {
|
||||
await globalPrismaClient.emailOutbox.createMany({
|
||||
data: options.recipients.map(recipient => ({
|
||||
tenancyId: options.tenancy.id,
|
||||
tsxSource: options.tsxSource,
|
||||
themeId: options.themeId,
|
||||
isHighPriority: options.isHighPriority,
|
||||
createdWith: options.createdWith.type === "draft" ? EmailOutboxCreatedWith.DRAFT : EmailOutboxCreatedWith.PROGRAMMATIC_CALL,
|
||||
emailDraftId: options.createdWith.type === "draft" ? options.createdWith.draftId : undefined,
|
||||
emailProgrammaticCallTemplateId: options.createdWith.type === "programmatic-call" ? options.createdWith.templateId : undefined,
|
||||
to: serializeRecipient(recipient)!,
|
||||
extraRenderVariables: options.extraVariables,
|
||||
scheduledAt: options.scheduledAt,
|
||||
shouldSkipDeliverabilityCheck: options.shouldSkipDeliverabilityCheck,
|
||||
overrideSubject: options.overrideSubject,
|
||||
overrideNotificationCategoryId: options.overrideNotificationCategoryId,
|
||||
})),
|
||||
});
|
||||
/** Stable IDs make retries idempotent. When provided, there must be exactly one ID per recipient. */
|
||||
emailOutboxIds?: string[],
|
||||
}): Promise<SendEmailToManyResult> {
|
||||
const rows = options.recipients.map((recipient) => ({
|
||||
tenancyId: options.tenancy.id,
|
||||
tsxSource: options.tsxSource,
|
||||
themeId: options.themeId,
|
||||
isHighPriority: options.isHighPriority,
|
||||
createdWith: options.createdWith.type === "draft" ? EmailOutboxCreatedWith.DRAFT : EmailOutboxCreatedWith.PROGRAMMATIC_CALL,
|
||||
emailDraftId: options.createdWith.type === "draft" ? options.createdWith.draftId : undefined,
|
||||
emailProgrammaticCallTemplateId: options.createdWith.type === "programmatic-call" ? options.createdWith.templateId : undefined,
|
||||
to: serializeRecipient(recipient)!,
|
||||
extraRenderVariables: options.extraVariables,
|
||||
scheduledAt: options.scheduledAt,
|
||||
shouldSkipDeliverabilityCheck: options.shouldSkipDeliverabilityCheck,
|
||||
overrideSubject: options.overrideSubject,
|
||||
overrideNotificationCategoryId: options.overrideNotificationCategoryId,
|
||||
}));
|
||||
const createResult = options.emailOutboxIds === undefined
|
||||
? await globalPrismaClient.emailOutbox.createMany({ data: rows })
|
||||
: await globalPrismaClient.emailOutbox.createMany(addIdempotencyToEmailOutboxRows(rows, options.emailOutboxIds));
|
||||
|
||||
if (!getEnvBoolean("STACK_EMAIL_BRANCHING_DISABLE_QUEUE_AUTO_TRIGGER")) {
|
||||
// The cron job should run runEmailQueueStep() to process the emails, but we call it here again for those self-hosters
|
||||
// who didn't set up the cron job correctly, and also just in case something happens to the cron job.
|
||||
runAsynchronouslyAndWaitUntil(runEmailQueueStep());
|
||||
}
|
||||
|
||||
return {
|
||||
createdCount: createResult.count,
|
||||
alreadyEnqueuedCount: options.emailOutboxIds === undefined ? 0 : options.recipients.length - createResult.count,
|
||||
};
|
||||
}
|
||||
|
||||
export async function sendEmailFromDefaultTemplate(options: {
|
||||
@ -225,3 +259,33 @@ import.meta.vitest?.test('normalizeEmail(...)', async ({ expect }) => {
|
||||
expect(() => normalizeEmail('test@multiple@domains.com')).toThrow();
|
||||
expect(() => normalizeEmail('invalid.email')).toThrow();
|
||||
});
|
||||
|
||||
import.meta.vitest?.describe('sendEmailToMany idempotency rows', () => {
|
||||
const { expect, test } = import.meta.vitest!;
|
||||
test('uses stable IDs and skipDuplicates when IDs are provided', () => {
|
||||
expect(addIdempotencyToEmailOutboxRows(
|
||||
[{ recipient: 'one' }, { recipient: 'two' }],
|
||||
['00000000-0000-4000-8000-000000000001', '00000000-0000-4000-8000-000000000002'],
|
||||
)).toEqual({
|
||||
data: [
|
||||
{ recipient: 'one', id: '00000000-0000-4000-8000-000000000001' },
|
||||
{ recipient: 'two', id: '00000000-0000-4000-8000-000000000002' },
|
||||
],
|
||||
skipDuplicates: true,
|
||||
});
|
||||
});
|
||||
|
||||
test('preserves existing non-idempotent inserts when IDs are omitted', () => {
|
||||
expect(addIdempotencyToEmailOutboxRows([{ recipient: 'one' }], undefined)).toEqual({
|
||||
data: [{ recipient: 'one' }],
|
||||
});
|
||||
});
|
||||
|
||||
test('rejects mismatched or duplicated IDs before insertion', () => {
|
||||
expect(() => addIdempotencyToEmailOutboxRows([{ recipient: 'one' }], [])).toThrow(/exactly one emailOutboxId/);
|
||||
expect(() => addIdempotencyToEmailOutboxRows(
|
||||
[{ recipient: 'one' }, { recipient: 'two' }],
|
||||
['00000000-0000-4000-8000-000000000001', '00000000-0000-4000-8000-000000000001'],
|
||||
)).toThrow(/unique emailOutboxIds/);
|
||||
});
|
||||
});
|
||||
|
||||
@ -6,7 +6,7 @@
|
||||
|
||||
import * as yup from "yup";
|
||||
import { ALL_APPS } from "../apps/apps-config";
|
||||
import { DEFAULT_EMAIL_TEMPLATES, DEFAULT_EMAIL_THEMES, DEFAULT_EMAIL_THEME_ID } from "../helpers/emails";
|
||||
import { DEFAULT_EMAIL_TEMPLATES, DEFAULT_EMAIL_THEMES, DEFAULT_EMAIL_THEME_ID, DEFAULT_TEMPLATE_IDS } from "../helpers/emails";
|
||||
import * as schemaFields from "../schema-fields";
|
||||
import { productSchema, userSpecifiedIdSchema, yupBoolean, yupDate, yupMixed, yupNever, yupNumber, yupObject, yupRecord, yupString, yupTuple, yupUnion } from "../schema-fields";
|
||||
import { SUPPORTED_CURRENCIES } from "../utils/currency-constants";
|
||||
@ -1124,10 +1124,11 @@ 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({
|
||||
const template = organizationConfigDefaults.emails.templates[DEFAULT_TEMPLATE_IDS.usage_email];
|
||||
expect(template).toMatchObject({
|
||||
displayName: "Usage Email",
|
||||
themeId: undefined,
|
||||
});
|
||||
expect(template.themeId).toBeUndefined();
|
||||
});
|
||||
|
||||
type _DeepOmitDefaultsImpl<T, U> = T extends object ? (
|
||||
|
||||
@ -254,4 +254,5 @@ export const DEFAULT_TEMPLATE_IDS = {
|
||||
sign_in_invitation: EMAIL_TEMPLATE_SIGN_IN_INVITATION_ID,
|
||||
payment_receipt: EMAIL_TEMPLATE_PAYMENT_RECEIPT_ID,
|
||||
payment_failed: EMAIL_TEMPLATE_PAYMENT_FAILED_ID,
|
||||
usage_email: EMAIL_TEMPLATE_USAGE_EMAIL_ID,
|
||||
} as const;
|
||||
|
||||
Loading…
Reference in New Issue
Block a user