feat(websub): wire real-time push notifications on email ingest

Pass ExecutionContext through the email processing chain so notifySubscribers
is called via ctx.waitUntil after a new email is stored.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/lib/cloudflare-email.ts

@@ -5,7 +5,7 @@ import { processEmail, RawAttachment } from "./email-processor";
 export async function handleCloudflareEmail(
   message: ForwardableEmailMessage,
   env: Env,
-  _ctx: ExecutionContext,
+  ctx: ExecutionContext,
 ): Promise<void> {
   try {
     const email = await PostalMime.parse(message.raw);
@@ -41,6 +41,7 @@ export async function handleCloudflareEmail(
         attachments: rawAttachments,
       },
       env,
+      ctx,
     );
   } catch (error) {
     console.error("Error processing Cloudflare email:", error);

src/lib/email-processor.ts

@@ -6,6 +6,7 @@ import {
   FeedConfig,
   FeedMetadata,
 } from "../types";
+import { notifySubscribers } from "../utils/websub";
 
 export interface RawAttachment {
   filename: string;
@@ -73,6 +74,7 @@ async function uploadAttachments(
 export async function processEmail(
   input: ProcessEmailInput,
   env: Env,
+  ctx?: ExecutionContext,
 ): Promise<Response> {
   const feedId = EmailParser.extractFeedId(input.toAddress);
   if (!feedId) {
@@ -182,5 +184,8 @@ export async function processEmail(
   ]);
 
   console.log(`Successfully processed email for feed ${feedId}`);
+  if (ctx) {
+    ctx.waitUntil(notifySubscribers(feedId, env));
+  }
   return new Response("Email processed successfully", { status: 200 });
 }

src/lib/forwardemail.ts

@@ -68,6 +68,7 @@ function toArrayBuffer(
 export async function handleForwardEmail(
   payload: ForwardEmailPayload,
   env: Env,
+  ctx?: ExecutionContext,
 ): Promise<Response> {
   const emailData = EmailParser.parseForwardEmailPayload(payload);
 
@@ -95,5 +96,6 @@ export async function handleForwardEmail(
       attachments: rawAttachments,
     },
     env,
+    ctx,
   );
 }

src/routes/inbound.ts

@@ -14,7 +14,7 @@ export async function handle(c: Context): Promise<Response> {
       contentType: payload.html ? "HTML" : "Text",
     });
 
-    return handleForwardEmail(payload, env);
+    return handleForwardEmail(payload, env, c.executionCtx);
   } catch (error) {
     console.error("Error processing email:", error);
     return new Response("Error processing email", { status: 500 });