Add custom sender email support with SES verification

Allow organizations to configure verified sender email addresses for the Send Email node via AWS SES. Includes API routes for setting, verifying, and removing sender emails, a new EmailService abstraction in the runtime, database migration, and frontend management UI.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: bchapuis

apps/api/.dev.vars.example

@@ -2,6 +2,7 @@ WEB_HOST=http://localhost:3001
 WEBSITE_URL=http://localhost:3002
 
 EMAIL_DOMAIN=dafthunk.com
+WORKFLOW_EMAIL_DOMAIN=mail.dafthunk.com
 
 CLOUDFLARE_ENV=development
 CLOUDFLARE_ACCOUNT_ID=CHANGE_ME

apps/api/src/context.ts

@@ -28,7 +28,7 @@ export interface Bindings {
   EXECUTIONS: AnalyticsEngineDataset;
   WEB_HOST: string;
   WEBSITE_URL: string;
-  EMAIL_DOMAIN: string;
+  WORKFLOW_EMAIL_DOMAIN: string;
   JWT_SECRET: string;
   CLOUDFLARE_ENV: string;
   CLOUDFLARE_ACCOUNT_ID: string;

apps/api/src/db/migrations/0049_add_sender_email.sql

@@ -0,0 +1,2 @@
+ALTER TABLE `emails` ADD COLUMN `sender_email` text;
+ALTER TABLE `emails` ADD COLUMN `sender_email_status` text;

apps/api/src/db/queries.ts

@@ -59,6 +59,7 @@ import {
   type SchemaInsert,
   type SchemaRow,
   type SecretInsert,
+  type SenderEmailStatusType,
   scheduledTriggers,
   schemas,
   secrets,
@@ -863,6 +864,8 @@ export async function getEmails(
     .select({
       id: emails.id,
       name: emails.name,
+      senderEmail: emails.senderEmail,
+      senderEmailStatus: emails.senderEmailStatus,
       createdAt: emails.createdAt,
       updatedAt: emails.updatedAt,
     })
@@ -2169,6 +2172,75 @@ export async function getOrganizationBillingInfo(
   return organization;
 }
 
+/**
+ * Get the sender email configuration for an email
+ */
+export async function getEmailSenderEmail(
+  db: ReturnType<typeof createDatabase>,
+  emailId: string,
+  organizationId: string
+): Promise<
+  | {
+      senderEmail: string | null;
+      senderEmailStatus: SenderEmailStatusType | null;
+    }
+  | undefined
+> {
+  const [email] = await db
+    .select({
+      senderEmail: emails.senderEmail,
+      senderEmailStatus: emails.senderEmailStatus,
+    })
+    .from(emails)
+    .where(
+      and(eq(emails.id, emailId), eq(emails.organizationId, organizationId))
+    )
+    .limit(1);
+  return email;
+}
+
+/**
+ * Update the sender email and status for an email
+ */
+export async function updateEmailSenderEmail(
+  db: ReturnType<typeof createDatabase>,
+  emailId: string,
+  organizationId: string,
+  senderEmailAddress: string,
+  status: SenderEmailStatusType
+) {
+  await db
+    .update(emails)
+    .set({
+      senderEmail: senderEmailAddress,
+      senderEmailStatus: status,
+      updatedAt: new Date(),
+    })
+    .where(
+      and(eq(emails.id, emailId), eq(emails.organizationId, organizationId))
+    );
+}
+
+/**
+ * Clear the sender email configuration for an email
+ */
+export async function clearEmailSenderEmail(
+  db: ReturnType<typeof createDatabase>,
+  emailId: string,
+  organizationId: string
+) {
+  await db
+    .update(emails)
+    .set({
+      senderEmail: null,
+      senderEmailStatus: null,
+      updatedAt: new Date(),
+    })
+    .where(
+      and(eq(emails.id, emailId), eq(emails.organizationId, organizationId))
+    );
+}
+
 /**
  * Derive user plan from organization billing info.
  * Pro if has active subscription OR canceled but still in billing period.
@@ -2851,6 +2923,31 @@ export async function isOrganizationOwner(
   return !!membership;
 }
 
+/**
+ * Check if a user is an admin or owner of an organization
+ */
+export async function isOrganizationAdminOrOwner(
+  db: ReturnType<typeof createDatabase>,
+  organizationId: string,
+  userId: string
+): Promise<boolean> {
+  const [membership] = await db
+    .select({ role: memberships.role })
+    .from(memberships)
+    .where(
+      and(
+        eq(memberships.userId, userId),
+        eq(memberships.organizationId, organizationId)
+      )
+    )
+    .limit(1);
+
+  return (
+    membership?.role === OrganizationRole.OWNER ||
+    membership?.role === OrganizationRole.ADMIN
+  );
+}
+
 /**
  * Add or update a user's membership in an organization
  *

apps/api/src/db/schema/index.ts

@@ -110,6 +110,16 @@ export const SubscriptionStatus = {
 export type SubscriptionStatusType =
   (typeof SubscriptionStatus)[keyof typeof SubscriptionStatus];
 
+// Sender email verification status
+export const SenderEmailStatus = {
+  PENDING: "pending",
+  VERIFIED: "verified",
+  FAILED: "failed",
+} as const;
+
+export type SenderEmailStatusType =
+  (typeof SenderEmailStatus)[keyof typeof SenderEmailStatus];
+
 /**
  * REUSABLE COLUMNS
  */
@@ -468,7 +478,7 @@ export const queueTriggers = sqliteTable(
   ]
 );
 
-// Emails - Email inboxes associated with organizations
+// Emails - Emails associated with organizations
 export const emails = sqliteTable(
   "emails",
   {
@@ -477,6 +487,10 @@ export const emails = sqliteTable(
     organizationId: text("organization_id")
       .notNull()
       .references(() => organizations.id, { onDelete: "cascade" }),
+    senderEmail: text("sender_email"),
+    senderEmailStatus: text(
+      "sender_email_status"
+    ).$type<SenderEmailStatusType>(),
     createdAt: createCreatedAt(),
     updatedAt: createUpdatedAt(),
   },
@@ -487,7 +501,7 @@ export const emails = sqliteTable(
   ]
 );
 
-// Email Triggers - Email inbox triggers for workflows
+// Email Triggers - Email triggers for workflows
 export const emailTriggers = sqliteTable(
   "email_triggers",
   {

apps/api/src/email.ts

@@ -51,16 +51,16 @@ export async function handleIncomingEmail(
     return;
   }
 
-  console.log(`Processing email trigger for email inbox: ${emailId}`);
+  console.log(`Processing email trigger for email: ${emailId}`);
 
   const db = createDatabase(env.DB);
   const workflowStore = new WorkflowStore(env);
 
-  // Get email inbox (globally unique UUID, org derived from record)
+  // Get email (globally unique UUID, org derived from record)
   const { getEmailById, getEmailTriggersByEmail } = await import("./db");
   const email = await getEmailById(db, emailId);
   if (!email) {
-    console.error(`Email inbox '${emailId}' not found`);
+    console.error(`Email '${emailId}' not found`);
     return;
   }
 
@@ -74,12 +74,12 @@ export async function handleIncomingEmail(
   );
 
   if (emailTriggersWithWorkflows.length === 0) {
-    console.log(`No active workflows found for email inbox: ${emailId}`);
+    console.log(`No active workflows found for email: ${emailId}`);
     return;
   }
 
   console.log(
-    `Found ${emailTriggersWithWorkflows.length} workflow(s) to trigger for email inbox ${emailId}`
+    `Found ${emailTriggersWithWorkflows.length} workflow(s) to trigger for email ${emailId}`
   );
 
   // Read raw email content once (stream can only be consumed once)