From d0764ddd8eca790cb0740d6066dbf22c39b0c339 Mon Sep 17 00:00:00 2001
From: Julien Herr <julien@herr.fr>
Date: Thu, 21 May 2026 23:08:19 +0200
Subject: [PATCH] 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 | 3 ++-
 src/lib/email-processor.ts  | 5 +++++
 src/lib/forwardemail.ts     | 2 ++
 src/routes/inbound.ts       | 2 +-
 4 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/src/lib/cloudflare-email.ts b/src/lib/cloudflare-email.ts
index a67d2fe..d0fd7b4 100644
--- a/src/lib/cloudflare-email.ts
+++ b/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);
diff --git a/src/lib/email-processor.ts b/src/lib/email-processor.ts
index 2a2ae97..2811eda 100644
--- a/src/lib/email-processor.ts
+++ b/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 });
 }
diff --git a/src/lib/forwardemail.ts b/src/lib/forwardemail.ts
index 2036f33..e11eae7 100644
--- a/src/lib/forwardemail.ts
+++ b/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,
   );
 }
diff --git a/src/routes/inbound.ts b/src/routes/inbound.ts
index cc0d69b..8aebfe8 100644
--- a/src/routes/inbound.ts
+++ b/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 });