diff --git a/apps/mail/app/(routes)/settings/privacy/page.tsx b/apps/mail/app/(routes)/settings/privacy/page.tsx
index 3ae49d0173..b95fafaa67 100644
--- a/apps/mail/app/(routes)/settings/privacy/page.tsx
+++ b/apps/mail/app/(routes)/settings/privacy/page.tsx
@@ -15,11 +15,11 @@ import { useTRPC } from '@/providers/query-provider';
 import { useMutation } from '@tanstack/react-query';
 // import { saveUserSettings } from '@/actions/settings';
 import { useSettings } from '@/hooks/use-settings';
-import { Button } from '@/components/ui/button';
 import { Switch } from '@/components/ui/switch';
+import { Button } from '@/components/ui/button';
 import { useState, useEffect } from 'react';
-import { m } from '@/paraglide/messages';
 import { useForm } from 'react-hook-form';
+import { m } from '@/paraglide/messages';
 import { XIcon } from 'lucide-react';
 import { toast } from 'sonner';
 import * as z from 'zod';
@@ -32,17 +32,16 @@ export default function PrivacyPage() {
 
   const form = useForm<z.infer<typeof userSettingsSchema>>({
     resolver: zodResolver(userSettingsSchema),
-    defaultValues: {
-      externalImages: true,
-      trustedSenders: [],
-    },
   });
 
-  const externalImages = form.watch('externalImages');
-
+  const externalImages = data?.settings.externalImages;
   useEffect(() => {
     if (data) {
-      form.reset(data.settings);
+      form.reset({
+        ...data.settings,
+        trustedSenders: data.settings.trustedSenders,
+        externalImages: !!data.settings.externalImages,
+      });
     }
   }, [form, data]);
 
@@ -87,10 +86,10 @@ export default function PrivacyPage() {
                   <FormItem className="bg-popover flex w-full flex-row items-center justify-between rounded-lg border p-4 md:w-auto">
                     <div className="space-y-0.5">
                       <FormLabel className="text-base">
-                      {m['pages.settings.privacy.externalImages']()}
+                        {m['pages.settings.privacy.externalImages']()}
                       </FormLabel>
                       <FormDescription>
-                      {m['pages.settings.privacy.externalImagesDescription']()}
+                        {m['pages.settings.privacy.externalImagesDescription']()}
                       </FormDescription>
                     </div>
                     <FormControl className="ml-4">
@@ -107,10 +106,10 @@ export default function PrivacyPage() {
                     <FormItem className="bg-popover flex w-full flex-col rounded-lg border p-4">
                       <div className="space-y-0.5">
                         <FormLabel className="text-base">
-                        {m['pages.settings.privacy.trustedSenders']()}
+                          {m['pages.settings.privacy.trustedSenders']()}
                         </FormLabel>
                         <FormDescription>
-                        {m['pages.settings.privacy.trustedSendersDescription']()}
+                          {m['pages.settings.privacy.trustedSendersDescription']()}
                         </FormDescription>
                       </div>
                       <ScrollArea className="flex max-h-32 flex-col pr-3">
diff --git a/apps/mail/components/mail/mail-content.tsx b/apps/mail/components/mail/mail-content.tsx
new file mode 100644
index 0000000000..9f4c02b0f8
--- /dev/null
+++ b/apps/mail/components/mail/mail-content.tsx
@@ -0,0 +1,206 @@
+import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
+import { useEffect, useMemo, useRef, useState, useCallback } from 'react';
+import { defaultUserSettings } from '@zero/server/schemas';
+import { fixNonReadableColors } from '@/lib/email-utils';
+import { useTRPC } from '@/providers/query-provider';
+import { getBrowserTimezone } from '@/lib/timezones';
+import { useSettings } from '@/hooks/use-settings';
+import { m } from '@/paraglide/messages';
+import { useTheme } from 'next-themes';
+import { cn } from '@/lib/utils';
+import { toast } from 'sonner';
+
+interface MailContentProps {
+  id: string;
+  html: string;
+  senderEmail: string;
+}
+
+export function MailContent({ id, html, senderEmail }: MailContentProps) {
+  const { data, refetch } = useSettings();
+  const queryClient = useQueryClient();
+  const isTrustedSender = useMemo(
+    () => data?.settings?.externalImages || data?.settings?.trustedSenders?.includes(senderEmail),
+    [data?.settings, senderEmail],
+  );
+  const [cspViolation, setCspViolation] = useState(false);
+  const [temporaryImagesEnabled, setTemporaryImagesEnabled] = useState(false);
+  const hostRef = useRef<HTMLDivElement>(null);
+  const shadowRootRef = useRef<ShadowRoot | null>(null);
+  const { resolvedTheme } = useTheme();
+  const trpc = useTRPC();
+
+  const { mutateAsync: saveUserSettings } = useMutation({
+    ...trpc.settings.save.mutationOptions(),
+    onSuccess: () => {
+      queryClient.invalidateQueries({ queryKey: ['settings'] });
+    },
+  });
+
+  const { mutateAsync: trustSender } = useMutation({
+    mutationFn: async () => {
+      const existingSettings = data?.settings ?? {
+        ...defaultUserSettings,
+        timezone: getBrowserTimezone(),
+      };
+
+      const { success } = await saveUserSettings({
+        ...existingSettings,
+        trustedSenders: data?.settings?.trustedSenders
+          ? data.settings.trustedSenders.concat(senderEmail)
+          : [senderEmail],
+      });
+
+      if (!success) {
+        throw new Error('Failed to trust sender');
+      }
+    },
+    onSuccess: () => {
+      refetch();
+    },
+    onError: () => {
+      toast.error('Failed to trust sender');
+    },
+  });
+
+  const { mutateAsync: processEmailContent } = useMutation(
+    trpc.mail.processEmailContent.mutationOptions(),
+  );
+
+  const { data: processedData } = useQuery({
+    queryKey: ['email-content', id, isTrustedSender || temporaryImagesEnabled, resolvedTheme],
+    queryFn: async () => {
+      const result = await processEmailContent({
+        html,
+        shouldLoadImages: isTrustedSender || temporaryImagesEnabled,
+        theme: (resolvedTheme as 'light' | 'dark') || 'light',
+      });
+
+      return {
+        html: result.processedHtml,
+        hasBlockedImages: result.hasBlockedImages,
+      };
+    },
+    staleTime: 30 * 60 * 1000,
+    gcTime: 60 * 60 * 1000,
+    refetchOnWindowFocus: false,
+    refetchOnMount: false,
+  });
+
+  useEffect(() => {
+    if (processedData) {
+      if (processedData.hasBlockedImages) {
+        setCspViolation(true);
+      }
+    }
+  }, [processedData]);
+
+  useEffect(() => {
+    if (!hostRef.current || shadowRootRef.current) return;
+
+    shadowRootRef.current = hostRef.current.attachShadow({ mode: 'open' });
+  }, []);
+
+  useEffect(() => {
+    if (!shadowRootRef.current || !processedData) return;
+
+    shadowRootRef.current.innerHTML = processedData.html;
+  }, [processedData]);
+
+  useEffect(() => {
+    if (!shadowRootRef.current) return;
+
+    const root = shadowRootRef.current;
+
+    const applyFix: () => void = () => {
+      const topLevelEls = Array.from(root.children) as HTMLElement[];
+      topLevelEls.forEach((el) => {
+        try {
+          fixNonReadableColors(el, {
+            defaultBackground: resolvedTheme === 'dark' ? 'rgb(10,10,10)' : '#ffffff',
+          });
+        } catch (err) {
+          console.error('Failed to fix colors in email content:', err);
+        }
+      });
+    };
+
+    requestAnimationFrame(applyFix);
+  }, [processedData, resolvedTheme]);
+
+  useEffect(() => {
+    if (isTrustedSender || temporaryImagesEnabled) {
+      setCspViolation(false);
+    }
+  }, [isTrustedSender, temporaryImagesEnabled]);
+
+  const handleImageError = useCallback(
+    (e: Event) => {
+      const target = e.target as HTMLImageElement;
+      if (target.tagName === 'IMG') {
+        if (!(isTrustedSender || temporaryImagesEnabled)) {
+          setCspViolation(true);
+        }
+        target.style.display = 'none';
+      }
+    },
+    [isTrustedSender, temporaryImagesEnabled],
+  );
+
+  useEffect(() => {
+    if (!shadowRootRef.current) return;
+
+    shadowRootRef.current.addEventListener('error', handleImageError, true);
+
+    const handleClick = (e: Event) => {
+      const target = e.target as HTMLElement;
+      if (target.tagName === 'A') {
+        e.preventDefault();
+        const href = target.getAttribute('href');
+        if (href && (href.startsWith('http://') || href.startsWith('https://'))) {
+          window.open(href, '_blank', 'noopener,noreferrer');
+        } else if (href && href.startsWith('mailto:')) {
+          window.location.href = href;
+        }
+      }
+    };
+
+    shadowRootRef.current.addEventListener('click', handleClick);
+
+    return () => {
+      shadowRootRef.current?.removeEventListener('error', handleImageError, true);
+      shadowRootRef.current?.removeEventListener('click', handleClick);
+    };
+  }, [handleImageError, processedData]);
+
+  return (
+    <>
+      {cspViolation && !isTrustedSender && !data?.settings?.externalImages && (
+        <div className="flex items-center justify-start bg-amber-600/20 px-2 py-1 text-sm text-amber-600">
+          <p>{m['common.actions.hiddenImagesWarning']()}</p>
+          <button
+            onClick={() => setTemporaryImagesEnabled(!temporaryImagesEnabled)}
+            className="ml-2 cursor-pointer underline"
+          >
+            {temporaryImagesEnabled
+              ? m['common.actions.disableImages']()
+              : m['common.actions.showImages']()}
+          </button>
+          <button
+            onClick={async () => {
+              try {
+                await trustSender();
+              } catch (error) {
+                console.error('Error trusting sender:', error);
+              }
+            }}
+            className="ml-2 cursor-pointer underline"
+          >
+            {m['common.actions.trustSender']()}
+          </button>
+        </div>
+      )}
+      <div ref={hostRef} className={cn('mail-content w-full flex-1 overflow-hidden')} />
+    </>
+  );
+}
diff --git a/apps/mail/components/mail/mail-display.tsx b/apps/mail/components/mail/mail-display.tsx
index b910af69dc..53306ced90 100644
--- a/apps/mail/components/mail/mail-display.tsx
+++ b/apps/mail/components/mail/mail-display.tsx
@@ -47,7 +47,7 @@ import { Markdown } from '@react-email/components';
 import { useSummary } from '@/hooks/use-summary';
 import { TextShimmer } from '../ui/text-shimmer';
 import { RenderLabels } from './render-labels';
-import { MailIframe } from './mail-iframe';
+import { MailContent } from './mail-content';
 import { m } from '@/paraglide/messages';
 import { useParams } from 'react-router';
 import { FileText } from 'lucide-react';
@@ -1205,7 +1205,7 @@ const MailDisplay = ({ emailData, index, totalEmails, demo, threadAttachments }:
             <!-- Email Body -->
             <div class="email-body">
               <div class="email-content">
-                ${escapeHtml(emailData.decodedBody) || '<p><em>No email content available</em></p>'}
+                ${escapeHtml(emailData?.decodedBody || '') || '<p><em>No email content available</em></p>'}
               </div>
             </div>
 
@@ -1768,7 +1768,11 @@ const MailDisplay = ({ emailData, index, totalEmails, demo, threadAttachments }:
               <div className="h-fit w-full p-0">
                 {/* mail main body */}
                 {emailData?.decodedBody ? (
-                  <MailIframe html={emailData?.decodedBody} senderEmail={emailData.sender.email} />
+                  <MailContent
+                    id={emailData.id}
+                    html={emailData?.decodedBody}
+                    senderEmail={emailData.sender.email}
+                  />
                 ) : null}
                 {/* mail attachments */}
                 {emailData?.attachments && emailData?.attachments.length > 0 ? (
diff --git a/apps/mail/components/mail/mail-iframe.tsx b/apps/mail/components/mail/mail-iframe.tsx
deleted file mode 100644
index e29fd805a2..0000000000
--- a/apps/mail/components/mail/mail-iframe.tsx
+++ /dev/null
@@ -1,174 +0,0 @@
-import { addStyleTags, doesContainStyleTags, template } from '@/lib/email-utils.client';
-import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
-import { useEffect, useMemo, useRef, useState, useCallback } from 'react';
-import { defaultUserSettings } from '@zero/server/schemas';
-import { fixNonReadableColors } from '@/lib/email-utils';
-import { useTRPC } from '@/providers/query-provider';
-import { getBrowserTimezone } from '@/lib/timezones';
-import { useSettings } from '@/hooks/use-settings';
-import { useTheme } from 'next-themes';
-import { cn } from '@/lib/utils';
-import { toast } from 'sonner';
-import { m } from '@/paraglide/messages';
-
-export function MailIframe({ html, senderEmail }: { html: string; senderEmail: string }) {
-  const { data, refetch } = useSettings();
-  const queryClient = useQueryClient();
-  const isTrustedSender = useMemo(
-    () => data?.settings?.externalImages || data?.settings?.trustedSenders?.includes(senderEmail),
-    [data?.settings, senderEmail],
-  );
-  const [cspViolation, setCspViolation] = useState(false);
-  const [temporaryImagesEnabled, setTemporaryImagesEnabled] = useState(false);
-  const iframeRef = useRef<HTMLIFrameElement>(null);
-  const [height, setHeight] = useState(0);
-  const { resolvedTheme } = useTheme();
-  const trpc = useTRPC();
-
-  const { mutateAsync: saveUserSettings } = useMutation({
-    ...trpc.settings.save.mutationOptions(),
-    onSuccess: () => {
-      queryClient.invalidateQueries({ queryKey: ['settings'] });
-    },
-  });
-
-  const { mutateAsync: trustSender } = useMutation({
-    mutationFn: async () => {
-      const existingSettings = data?.settings ?? {
-        ...defaultUserSettings,
-        timezone: getBrowserTimezone(),
-      };
-
-      const { success } = await saveUserSettings({
-        ...existingSettings,
-        trustedSenders: data?.settings.trustedSenders
-          ? data.settings.trustedSenders.concat(senderEmail)
-          : [senderEmail],
-      });
-
-      if (!success) {
-        throw new Error('Failed to trust sender');
-      }
-    },
-    onSuccess: () => {
-      refetch();
-    },
-    onError: () => {
-      toast.error('Failed to trust sender');
-    },
-  });
-
-  const { data: processedHtml } = useQuery({
-    queryKey: ['email-template', html, isTrustedSender || temporaryImagesEnabled],
-    queryFn: () => template(html, isTrustedSender || temporaryImagesEnabled),
-    staleTime: 30 * 60 * 1000, // Increase cache time to 30 minutes
-    gcTime: 60 * 60 * 1000, // Keep in cache for 1 hour
-    refetchOnWindowFocus: false, // Don't refetch on window focus
-    refetchOnMount: false, // Don't refetch on mount if data exists
-  });
-
-
-
-  const calculateAndSetHeight = useCallback(() => {
-    if (!iframeRef.current?.contentWindow?.document.body) return;
-
-    const body = iframeRef.current.contentWindow.document.body;
-    const boundingRectHeight = body.getBoundingClientRect().height;
-    const scrollHeight = body.scrollHeight;
-
-    if (body.innerText.trim() === '') {
-      setHeight(0);
-      return;
-    }
-
-    // Use the larger of the two values to ensure all content is visible
-    const newHeight = Math.max(boundingRectHeight, scrollHeight);
-    setHeight(newHeight);
-  }, [iframeRef, setHeight]);
-
-  useEffect(() => {
-    if (!iframeRef.current || !processedHtml) return;
-
-    let finalHtml = processedHtml;
-    const containsStyleTags = doesContainStyleTags(processedHtml);
-    if (!containsStyleTags) {
-      finalHtml = addStyleTags(processedHtml);
-    }
-
-    const url = URL.createObjectURL(new Blob([finalHtml], { type: 'text/html' }));
-    iframeRef.current.src = url;
-
-    const handler = () => {
-      if (iframeRef.current?.contentWindow?.document.body) {
-        calculateAndSetHeight();
-        fixNonReadableColors(iframeRef.current.contentWindow.document.body);
-      }
-      setTimeout(calculateAndSetHeight, 500);
-    };
-
-    iframeRef.current.onload = handler;
-
-    return () => {
-      URL.revokeObjectURL(url);
-    };
-  }, [processedHtml, calculateAndSetHeight]);
-
-  useEffect(() => {
-    if (iframeRef.current?.contentWindow?.document.body) {
-      const body = iframeRef.current.contentWindow.document.body;
-      body.style.backgroundColor =
-        resolvedTheme === 'dark' ? 'rgb(10, 10, 10)' : 'rgb(245, 245, 245)';
-      requestAnimationFrame(() => {
-        fixNonReadableColors(body);
-      });
-    }
-  }, [resolvedTheme]);
-
-  useEffect(() => {
-    const ctrl = new AbortController();
-    window.addEventListener(
-      'message',
-      (event) => {
-        if (event.data.type === 'csp-violation') {
-          setCspViolation(true);
-        }
-      },
-      { signal: ctrl.signal },
-    );
-
-    return () => ctrl.abort();
-  }, []);
-
-  return (
-    <>
-      {cspViolation && !isTrustedSender && !data?.settings?.externalImages && (
-        <div className="flex items-center justify-start bg-amber-600/20 px-2 py-1 text-sm text-amber-600">
-          <p>{m['common.actions.hiddenImagesWarning']()}</p>
-          <button
-            onClick={() => setTemporaryImagesEnabled(!temporaryImagesEnabled)}
-            className="ml-2 cursor-pointer underline"
-          >
-            {temporaryImagesEnabled
-              ? m['common.actions.disableImages']()
-              : m['common.actions.showImages']()}
-          </button>
-          <button onClick={() => void trustSender()} className="ml-2 cursor-pointer underline">
-            {m['common.actions.trustSender']()}
-          </button>
-        </div>
-      )}
-      <iframe
-        height={height}
-        ref={iframeRef}
-        className={cn(
-          '!min-h-0 w-full flex-1 overflow-hidden px-4 transition-opacity duration-200',
-        )}
-        title="Email Content"
-        style={{
-          width: '100%',
-          overflow: 'hidden',
-        }}
-      />
-    </>
-  );
-}
diff --git a/apps/mail/hooks/use-mail-navigation.ts b/apps/mail/hooks/use-mail-navigation.ts
index f7c18f3d0b..2b4326527c 100644
--- a/apps/mail/hooks/use-mail-navigation.ts
+++ b/apps/mail/hooks/use-mail-navigation.ts
@@ -4,6 +4,7 @@ import { useOptimisticActions } from './use-optimistic-actions';
 import { useMail } from '@/components/mail/use-mail';
 import { useHotkeys } from 'react-hotkeys-hook';
 import { atom, useAtom } from 'jotai';
+import { useQueryState } from 'nuqs';
 
 export const focusedIndexAtom = atom<number | null>(null);
 export const mailNavigationCommandAtom = atom<null | 'next' | 'previous'>(null);
@@ -23,7 +24,8 @@ export function useMailNavigation({ items, containerRef, onNavigate }: UseMailNa
   itemsRef.current = items;
   const onNavigateRef = useRef(onNavigate);
   onNavigateRef.current = onNavigate;
-  const { open: isCommandPaletteOpen } = useCommandPalette();
+  const [threadId] = useQueryState('threadId');
+  const [isCommandPaletteOpen] = useQueryState('isCommandPaletteOpen');
 
   const hoveredMailRef = useRef<string | null>(null);
   const keyboardActiveRef = useRef(false);
@@ -77,8 +79,7 @@ export function useMailNavigation({ items, containerRef, onNavigate }: UseMailNa
       const message = itemsRef.current[index];
       const threadId = message.id;
 
-      const currentThreadId = window.location.search.includes('threadId=');
-      if (currentThreadId) {
+      if (threadId) {
         onNavigateRef.current(threadId);
         optimisticMarkAsRead([threadId], true);
       }
@@ -88,7 +89,7 @@ export function useMailNavigation({ items, containerRef, onNavigate }: UseMailNa
         bulkSelected: [],
       }));
     },
-    [setMail],
+    [setMail, threadId],
   );
 
   const navigateNext = useCallback(() => {
@@ -196,11 +197,14 @@ export function useMailNavigation({ items, containerRef, onNavigate }: UseMailNa
   }, [setFocusedIndex, onNavigateRef]);
 
   useHotkeys('ArrowUp', handleArrowUp, { preventDefault: true, enabled: !isCommandPaletteOpen });
-  useHotkeys('ArrowDown', handleArrowDown, { preventDefault: true, enabled: !isCommandPaletteOpen });
-  useHotkeys('j', handleArrowDown,{enabled: !isCommandPaletteOpen });
+  useHotkeys('ArrowDown', handleArrowDown, {
+    preventDefault: true,
+    enabled: !isCommandPaletteOpen,
+  });
+  useHotkeys('j', handleArrowDown, { enabled: !isCommandPaletteOpen });
   useHotkeys('k', handleArrowUp, { enabled: !isCommandPaletteOpen });
-  useHotkeys('Enter', handleEnter, { preventDefault: true,enabled: !isCommandPaletteOpen });
-  useHotkeys('Escape', handleEscape, { preventDefault: true,enabled: !isCommandPaletteOpen });
+  useHotkeys('Enter', handleEnter, { preventDefault: true, enabled: !isCommandPaletteOpen });
+  useHotkeys('Escape', handleEscape, { preventDefault: true, enabled: !isCommandPaletteOpen });
 
   const handleMouseEnter = useCallback(
     (threadId: string) => {
diff --git a/apps/mail/hooks/use-threads.ts b/apps/mail/hooks/use-threads.ts
index ea3e21a4a9..f524701597 100644
--- a/apps/mail/hooks/use-threads.ts
+++ b/apps/mail/hooks/use-threads.ts
@@ -1,20 +1,22 @@
+import { useInfiniteQuery, useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
 import { backgroundQueueAtom, isThreadInBackgroundQueueAtom } from '@/store/backgroundQueue';
-import { useInfiniteQuery, useQuery, useQueryClient } from '@tanstack/react-query';
 import type { IGetThreadResponse } from '../../server/src/lib/driver/types';
 import { useSearchValue } from '@/hooks/use-search-value';
 import { useTRPC } from '@/providers/query-provider';
 import useSearchLabels from './use-labels-search';
 import { useSession } from '@/lib/auth-client';
 import { useAtom, useAtomValue } from 'jotai';
+import { useSettings } from './use-settings';
 import { usePrevious } from './use-previous';
+import type { ParsedMessage } from '@/types';
 import { useEffect, useMemo } from 'react';
 import { useParams } from 'react-router';
+import { useTheme } from 'next-themes';
 import { useQueryState } from 'nuqs';
 
 export const useThreads = () => {
   const { folder } = useParams<{ folder: string }>();
   const [searchValue] = useSearchValue();
-  const { data: session } = useSession();
   const [backgroundQueue] = useAtom(backgroundQueueAtom);
   const isInQueue = useAtomValue(isThreadInBackgroundQueueAtom);
   const trpc = useTRPC();
@@ -67,6 +69,8 @@ export const useThread = (threadId: string | null, historyId?: string | null) =>
   const [_threadId] = useQueryState('threadId');
   const id = threadId ? threadId : _threadId;
   const trpc = useTRPC();
+  const { data } = useSettings();
+  const { resolvedTheme } = useTheme();
 
   const previousHistoryId = usePrevious(historyId ?? null);
   const queryClient = useQueryClient();
@@ -88,11 +92,45 @@ export const useThread = (threadId: string | null, historyId?: string | null) =>
     ),
   );
 
+  const isTrustedSender = useMemo(
+    () =>
+      !!data?.settings?.externalImages ||
+      !!data?.settings?.trustedSenders?.includes(threadQuery.data?.latest?.sender.email ?? ''),
+    [data?.settings, threadQuery.data?.latest?.sender.email],
+  );
+
   const latestDraft = useMemo(() => {
     if (!threadQuery.data?.latest?.id) return undefined;
     return threadQuery.data.messages.findLast((e) => e.isDraft);
   }, [threadQuery]);
 
+  const { mutateAsync: processEmailContent } = useMutation(
+    trpc.mail.processEmailContent.mutationOptions(),
+  );
+
+  const prefetchEmailContent = async (message: ParsedMessage) => {
+    return queryClient.prefetchQuery({
+      queryKey: ['email-content', message.id, isTrustedSender, resolvedTheme],
+      queryFn: async () => {
+        const result = await processEmailContent({
+          html: message.decodedBody ?? '',
+          shouldLoadImages: isTrustedSender,
+          theme: (resolvedTheme as 'light' | 'dark') || 'light',
+        });
+
+        return {
+          html: result.processedHtml,
+          hasBlockedImages: result.hasBlockedImages,
+        };
+      },
+    });
+  };
+
+  useEffect(() => {
+    if (!threadQuery.data?.latest?.id) return;
+    prefetchEmailContent(threadQuery.data.latest);
+  }, [threadQuery.data?.latest]);
+
   const isGroupThread = useMemo(() => {
     if (!threadQuery.data?.latest?.id) return false;
     const totalRecipients = [
diff --git a/apps/mail/lib/email-utils.ts b/apps/mail/lib/email-utils.ts
index 5fa8738e91..af84480ded 100644
--- a/apps/mail/lib/email-utils.ts
+++ b/apps/mail/lib/email-utils.ts
@@ -2,7 +2,11 @@ import * as emailAddresses from 'email-addresses';
 import type { Sender } from '@/types';
 import Color from 'color';
 
-export const fixNonReadableColors = (rootElement: HTMLElement, minContrast = 3.5) => {
+export const fixNonReadableColors = (
+  rootElement: HTMLElement,
+  options?: { minContrast?: number; defaultBackground?: string },
+) => {
+  const { minContrast = 3.5, defaultBackground = '#ffffff' } = options || {};
   const elements = Array.from<HTMLElement>(rootElement.querySelectorAll('*'));
   elements.unshift(rootElement);
 
@@ -21,7 +25,7 @@ export const fixNonReadableColors = (rootElement: HTMLElement, minContrast = 3.5
 
     try {
       const textColor = Color(style.color);
-      const effectiveBg = getEffectiveBackgroundColor(el);
+      const effectiveBg = getEffectiveBackgroundColor(el, defaultBackground);
 
       const blendedText =
         textColor.alpha() < 1 ? effectiveBg.mix(textColor, effectiveBg.alpha()) : textColor;
@@ -38,14 +42,14 @@ export const fixNonReadableColors = (rootElement: HTMLElement, minContrast = 3.5
   }
 };
 
-const getEffectiveBackgroundColor = (element: HTMLElement) => {
+const getEffectiveBackgroundColor = (element: HTMLElement, defaultBackground: string) => {
   let current: HTMLElement | null = element;
   while (current) {
     const bg = Color(getComputedStyle(current).backgroundColor);
     if (bg.alpha() >= 1) return bg.rgb();
     current = current.parentElement;
   }
-  return Color('#ffffff');
+  return Color(defaultBackground);
 };
 
 type ListUnsubscribeAction =
diff --git a/apps/server/src/lib/email-processor.ts b/apps/server/src/lib/email-processor.ts
new file mode 100644
index 0000000000..2ebd3fa1ca
--- /dev/null
+++ b/apps/server/src/lib/email-processor.ts
@@ -0,0 +1,136 @@
+import sanitizeHtml from 'sanitize-html';
+import * as cheerio from 'cheerio';
+
+interface ProcessEmailOptions {
+  html: string;
+  shouldLoadImages: boolean;
+  theme: 'light' | 'dark';
+}
+
+export function processEmailHtml({ html, shouldLoadImages, theme }: ProcessEmailOptions): {
+  processedHtml: string;
+  hasBlockedImages: boolean;
+} {
+  let hasBlockedImages = false;
+
+  const sanitizeConfig: sanitizeHtml.IOptions = {
+    allowedTags: sanitizeHtml.defaults.allowedTags.concat(['img']),
+    allowedAttributes: false,
+    allowedSchemes: shouldLoadImages
+      ? ['http', 'https', 'mailto', 'tel', 'data', 'cid', 'blob']
+      : ['http', 'https', 'mailto', 'tel', 'cid'],
+    allowedSchemesByTag: {
+      img: shouldLoadImages ? ['http', 'https', 'data', 'cid', 'blob'] : ['cid'],
+    },
+    transformTags: {
+      img: (tagName, attribs) => {
+        if (!shouldLoadImages && attribs.src && !attribs.src.startsWith('cid:')) {
+          hasBlockedImages = true;
+          return { tagName: 'span', attribs: { style: 'display:none;' } };
+        }
+        return { tagName, attribs };
+      },
+      a: (tagName, attribs) => {
+        return {
+          tagName,
+          attribs: {
+            ...attribs,
+            target: attribs.target || '_blank',
+            rel: 'noopener noreferrer',
+          },
+        };
+      },
+    },
+  };
+
+  const sanitized = sanitizeHtml(html, sanitizeConfig);
+
+  const $ = cheerio.load(sanitized);
+
+  $('img[width="1"][height="1"]').remove();
+  $('img[width="0"][height="0"]').remove();
+
+  $('.preheader, .preheaderText, [class*="preheader"]').each((_, el) => {
+    const $el = $(el);
+    const style = $el.attr('style') || '';
+    if (
+      style.includes('display:none') ||
+      style.includes('display: none') ||
+      style.includes('font-size:0') ||
+      style.includes('font-size: 0') ||
+      style.includes('line-height:0') ||
+      style.includes('line-height: 0') ||
+      style.includes('max-height:0') ||
+      style.includes('max-height: 0') ||
+      style.includes('mso-hide:all') ||
+      style.includes('opacity:0') ||
+      style.includes('opacity: 0')
+    ) {
+      $el.remove();
+    }
+  });
+
+  const minimalStyles = `
+    <style type="text/css">
+      :host {
+        display: block;
+        width: 100%;
+        height: 100%;
+        overflow: auto;
+        font-family: 'Geist', -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif !important;
+        line-height: 1.5;
+        background-color: ${theme === 'dark' ? '#1A1A1A' : '#ffffff'};
+        color: ${theme === 'dark' ? '#ffffff' : '#000000'};
+      }
+
+      *, *::before, *::after {
+        box-sizing: border-box;
+      }
+
+      body {
+        margin: 0;
+        padding: 0;
+      }
+
+      a {
+        cursor: pointer;
+        color: ${theme === 'dark' ? '#60a5fa' : '#2563eb'};
+        text-decoration: underline;
+      }
+
+      a:hover {
+        color: ${theme === 'dark' ? '#93bbfc' : '#1d4ed8'};
+      }
+
+      img {
+        max-width: 100%;
+        height: auto;
+        display: block;
+      }
+
+      table {
+        border-collapse: collapse;
+      }
+
+      .gmail_quote {
+        margin: 1em 0;
+        padding-left: 1em;
+        border-left: 1px solid ${theme === 'dark' ? '#666' : '#ccc'};
+      }
+
+      ::selection {
+        background: #b3d4fc;
+        text-shadow: none;
+      }
+    </style>
+  `;
+
+  const fullHtml = $.html();
+
+  const finalHtml = `${minimalStyles}${fullHtml}`;
+
+  return {
+    processedHtml: finalHtml,
+    hasBlockedImages,
+  };
+}
diff --git a/apps/server/src/pipelines.ts b/apps/server/src/pipelines.ts
index 6a19acf62e..a7712c70d1 100644
--- a/apps/server/src/pipelines.ts
+++ b/apps/server/src/pipelines.ts
@@ -33,7 +33,9 @@ const showLogs = true;
 const log = (message: string, ...args: any[]) => {
   if (showLogs) {
     console.log(message, ...args);
+    return message;
   }
+  return 'no message';
 };
 
 type VectorizeVectorMetadata = 'connection' | 'thread' | 'summary';
@@ -48,9 +50,14 @@ export class MainWorkflow extends WorkflowEntrypoint<Env, Params> {
     log('[MAIN_WORKFLOW] Starting workflow with payload:', event.payload);
     try {
       const { providerId, historyId, subscriptionName } = event.payload;
+
+      if (!env.GOOGLE_S_ACCOUNT) {
+        throw new Error('GOOGLE_S_ACCOUNT environment variable is not set');
+      }
+
       const serviceAccount = JSON.parse(env.GOOGLE_S_ACCOUNT);
       const connectionId = await step.do(
-        `[ZERO] Validate Arguments ${providerId} ${subscriptionName} ${historyId}`,
+        `[MAIN_WORKFLOW] Validate Arguments ${providerId} ${subscriptionName} ${historyId}`,
         async () => {
           log('[MAIN_WORKFLOW] Validating arguments');
           const regex = new RegExp(
@@ -78,44 +85,47 @@ export class MainWorkflow extends WorkflowEntrypoint<Env, Params> {
       const previousHistoryId = await env.gmail_history_id.get(connectionId);
       if (providerId === EProviders.google) {
         log('[MAIN_WORKFLOW] Processing Google provider workflow');
-        await step.do(`[ZERO] Send to Zero Workflow ${connectionId} ${historyId}`, async () => {
-          log('[MAIN_WORKFLOW] Previous history ID:', previousHistoryId);
-          if (previousHistoryId) {
-            log('[MAIN_WORKFLOW] Creating workflow instance with previous history');
-            const instance = await env.ZERO_WORKFLOW.create({
-              params: {
-                connectionId,
-                historyId: previousHistoryId,
-                nextHistoryId: historyId,
-              },
-            });
-            log('[MAIN_WORKFLOW] Created instance:', {
-              id: instance.id,
-              status: await instance.status(),
-            });
-          } else {
-            log('[MAIN_WORKFLOW] Creating workflow instance with current history');
-            const existingInstance = await env.ZERO_WORKFLOW.get(
-              `${connectionId}__${historyId}`,
-            ).catch(() => null);
-            if (existingInstance && (await existingInstance.status()).status === 'running') {
-              log('[MAIN_WORKFLOW] History already processing:', existingInstance.id);
-              return;
+        await step.do(
+          `[MAIN_WORKFLOW] Send to Zero Workflow ${connectionId} ${historyId}`,
+          async () => {
+            log('[MAIN_WORKFLOW] Previous history ID:', previousHistoryId);
+            if (previousHistoryId) {
+              log('[MAIN_WORKFLOW] Creating workflow instance with previous history');
+              const instance = await env.ZERO_WORKFLOW.create({
+                params: {
+                  connectionId,
+                  historyId: previousHistoryId,
+                  nextHistoryId: historyId,
+                },
+              });
+              log('[MAIN_WORKFLOW] Created instance:', {
+                id: instance.id,
+                status: await instance.status(),
+              });
+            } else {
+              log('[MAIN_WORKFLOW] Creating workflow instance with current history');
+              //   const existingInstance = await env.ZERO_WORKFLOW.get(
+              //     `${connectionId}__${historyId}`,
+              //   ).catch(() => null);
+              //   if (existingInstance && (await existingInstance.status()).status === 'running') {
+              //     log('[MAIN_WORKFLOW] History already processing:', existingInstance.id);
+              //     return;
+              //   }
+              const instance = await env.ZERO_WORKFLOW.create({
+                // id: `${connectionId}__${historyId}`,
+                params: {
+                  connectionId,
+                  historyId: historyId,
+                  nextHistoryId: historyId,
+                },
+              });
+              log('[MAIN_WORKFLOW] Created instance:', {
+                id: instance.id,
+                status: await instance.status(),
+              });
             }
-            const instance = await env.ZERO_WORKFLOW.create({
-              id: `${connectionId}__${historyId}`,
-              params: {
-                connectionId,
-                historyId: historyId,
-                nextHistoryId: historyId,
-              },
-            });
-            log('[MAIN_WORKFLOW] Created instance:', {
-              id: instance.id,
-              status: await instance.status(),
-            });
-          }
-        });
+          },
+        );
       } else {
         log('[MAIN_WORKFLOW] Unsupported provider:', providerId);
         throw new Error(`Unsupported provider: ${providerId}`);
@@ -146,36 +156,40 @@ export class ZeroWorkflow extends WorkflowEntrypoint<Env, Params> {
 
       const historyProcessingKey = `history_${connectionId}__${historyId}`;
       const isProcessing = await env.gmail_processing_threads.get(historyProcessingKey);
-      if (isProcessing) {
-        log('[ZERO_WORKFLOW] History already being processed:', {
+      if (isProcessing === 'true') {
+        return log('[ZERO_WORKFLOW] History already being processed:', {
           connectionId,
           historyId,
           processingStatus: isProcessing,
         });
-        return;
       }
 
-      await env.gmail_processing_threads.put(historyProcessingKey, 'true');
+      await env.gmail_processing_threads.put(historyProcessingKey, 'true', { expirationTtl: 3600 });
       log('[ZERO_WORKFLOW] Set processing flag for history:', historyProcessingKey);
+
       const { db, conn } = createDb(env.HYPERDRIVE.connectionString);
-      const foundConnection = await step.do(`[ZERO] Find Connection ${connectionId}`, async () => {
-        log('[ZERO_WORKFLOW] Finding connection:', connectionId);
-        const [foundConnection] = await db
-          .select()
-          .from(connection)
-          .where(eq(connection.id, connectionId.toString()));
-        if (!foundConnection) throw new Error(`Connection not found ${connectionId}`);
-        if (!foundConnection.accessToken || !foundConnection.refreshToken)
-          throw new Error(`Connection is not authorized ${connectionId}`);
-        log('[ZERO_WORKFLOW] Found connection:', foundConnection.id);
-        return foundConnection;
-      });
+
+      const foundConnection = await step.do(
+        `[ZERO_WORKFLOW] Find Connection ${connectionId}`,
+        async () => {
+          log('[ZERO_WORKFLOW] Finding connection:', connectionId);
+          const [foundConnection] = await db
+            .select()
+            .from(connection)
+            .where(eq(connection.id, connectionId.toString()));
+          if (!foundConnection) throw new Error(`Connection not found ${connectionId}`);
+          if (!foundConnection.accessToken || !foundConnection.refreshToken)
+            throw new Error(`Connection is not authorized ${connectionId}`);
+          log('[ZERO_WORKFLOW] Found connection:', foundConnection.id);
+          return foundConnection;
+        },
+      );
 
       const driver = connectionToDriver(foundConnection);
       if (foundConnection.providerId === EProviders.google) {
         log('[ZERO_WORKFLOW] Processing Google provider workflow');
         const history = await step.do(
-          `[ZERO] Get Gmail History for ${foundConnection.id}`,
+          `[ZERO_WORKFLOW] Get Gmail History ${foundConnection.id} ${historyId}`,
           async () => {
             try {
               log('[ZERO_WORKFLOW] Getting Gmail history with ID:', historyId);
@@ -195,47 +209,56 @@ export class ZeroWorkflow extends WorkflowEntrypoint<Env, Params> {
             }
           },
         );
-        await step.do(`[ZERO] Update next history id for ${foundConnection.id}`, async () => {
-          log('[ZERO_WORKFLOW] Updating next history ID:', nextHistoryId);
-          await env.gmail_history_id.put(connectionId.toString(), nextHistoryId.toString());
-        });
-        const threadsAdded = await step.do('[ZERO] Get new Threads', async () => {
-          log('[ZERO_WORKFLOW] Finding threads with changed messages');
-          const historiesWithChangedMessages = history.filter(
-            (history) => history.messagesAdded?.length,
-          );
-          const threadsAdded = [
-            ...new Set(
-              historiesWithChangedMessages.flatMap((history) =>
-                history
-                  .messagesAdded!.map((message) => message.message?.threadId)
-                  .filter((threadId): threadId is string => threadId !== undefined),
+        await step.do(
+          `[ZERO_WORKFLOW] Update next history id ${foundConnection.id} ${nextHistoryId}`,
+          async () => {
+            log('[ZERO_WORKFLOW] Updating next history ID:', nextHistoryId);
+            await env.gmail_history_id.put(connectionId.toString(), nextHistoryId.toString());
+          },
+        );
+        const threadsAdded = await step.do(
+          `[ZERO_WORKFLOW] Get new Threads ${connectionId}`,
+          async () => {
+            log('[ZERO_WORKFLOW] Finding threads with changed messages');
+            const historiesWithChangedMessages = history.filter(
+              (history) => history.messagesAdded?.length,
+            );
+            const threadsAdded = [
+              ...new Set(
+                historiesWithChangedMessages.flatMap((history) =>
+                  history
+                    .messagesAdded!.map((message) => message.message?.threadId)
+                    .filter((threadId): threadId is string => threadId !== undefined),
+                ),
               ),
-            ),
-          ];
-          log('[ZERO_WORKFLOW] Found new threads:', threadsAdded.length);
-          return threadsAdded;
-        });
+            ];
+            log('[ZERO_WORKFLOW] Found new threads:', threadsAdded.length);
+            return threadsAdded;
+          },
+        );
 
-        const threadsAddLabels = await step.do('[ZERO] Get Threads with new labels', async () => {
-          log('[ZERO_WORKFLOW] Finding threads with new labels');
-          const historiesWithNewLabels = history.filter((history) => history.labelsAdded?.length);
-          const threadsWithLabelsAdded = [
-            ...new Set(
-              historiesWithNewLabels.flatMap((history) =>
-                history
-                  .labelsAdded!.filter((label) => label.message?.threadId)
-                  .map((label) => label.message!.threadId)
-                  .filter((threadId): threadId is string => threadId !== undefined),
+        const threadsAddLabels = await step.do(
+          `[ZERO_WORKFLOW] Get Threads with new labels ${connectionId}`,
+          async () => {
+            log('[ZERO_WORKFLOW] Finding threads with new labels');
+            const historiesWithNewLabels = history.filter((history) => history.labelsAdded?.length);
+            const threadsWithLabelsAdded = [
+              ...new Set(
+                historiesWithNewLabels.flatMap((history) =>
+                  history
+                    .labelsAdded!.filter((label) => label.message?.threadId)
+                    .map((label) => label.message!.threadId)
+                    .filter((threadId): threadId is string => threadId !== undefined),
+                ),
               ),
-            ),
-          ];
-          log('[ZERO_WORKFLOW] Found threads with new labels:', threadsWithLabelsAdded.length);
-          return threadsWithLabelsAdded;
-        });
+            ];
+            log('[ZERO_WORKFLOW] Found threads with new labels:', threadsWithLabelsAdded.length);
+            return threadsWithLabelsAdded;
+          },
+        );
 
         const threadsRemoveLabels = await step.do(
-          '[ZERO] Get Threads with removed labels',
+          `[ZERO_WORKFLOW] Get Threads with removed labels ${connectionId}`,
           async () => {
             log('[ZERO_WORKFLOW] Finding threads with removed labels');
             const historiesWithRemovedLabels = history.filter(
@@ -259,85 +282,125 @@ export class ZeroWorkflow extends WorkflowEntrypoint<Env, Params> {
           },
         );
 
-        const lastPage = await step.do('[ZERO] Get last page', async () => {
-          log('[ZERO_WORKFLOW] Getting last page of threads');
-          const lastThreads = await driver.list({
-            folder: 'inbox',
-            query: 'NOT is:spam',
-            maxResults: 10,
-          });
-          log('[ZERO_WORKFLOW] Found threads in last page:', lastThreads.threads.length);
-          return lastThreads.threads.map((thread) => thread.id);
-        });
+        // const lastPage = await step.do(
+        //   `[ZERO_WORKFLOW] Get last page ${connectionId}`,
+        //   async () => {
+        //     log('[ZERO_WORKFLOW] Getting last page of threads');
+        //     const lastThreads = await driver.list({
+        //       folder: 'inbox',
+        //       query: 'NOT is:spam',
+        //       maxResults: 10,
+        //     });
+        //     log('[ZERO_WORKFLOW] Found threads in last page:', lastThreads.threads.length);
+        //     return lastThreads.threads.map((thread) => thread.id);
+        //   },
+        // );
 
-        const threadsToProcess = await step.do('[ZERO] Get threads to process', async () => {
-          log('[ZERO_WORKFLOW] Combining threads to process');
-          const threadsToProcess = [
-            ...new Set([...threadsAdded, ...lastPage, ...threadsAddLabels, ...threadsRemoveLabels]),
-          ];
-          log('[ZERO_WORKFLOW] Total threads to process:', threadsToProcess.length);
-          return threadsToProcess;
-        });
+        const threadsToProcess = await step.do(
+          `[ZERO_WORKFLOW] Get threads to process ${connectionId}`,
+          async () => {
+            log('[ZERO_WORKFLOW] Combining threads to process');
+            const threadsToProcess = [
+              ...new Set([
+                ...threadsAdded,
+                // ...lastPage,
+                ...threadsAddLabels,
+                ...threadsRemoveLabels,
+              ]),
+            ];
+            log('[ZERO_WORKFLOW] Total threads to process:', threadsToProcess.length);
+            return threadsToProcess;
+          },
+        );
 
-        await step.do(`[ZERO] Send Thread Workflow Instances`, async () => {
-          for (const threadId of threadsToProcess) {
-            try {
-              const isProcessing = await env.gmail_processing_threads.get(threadId.toString());
-              if (isProcessing) {
-                log('[ZERO_WORKFLOW] Thread already processing:', isProcessing, threadId);
-                continue;
-              }
-              await env.gmail_processing_threads.put(threadId.toString(), 'true');
-              const existingInstance = await env.THREAD_WORKFLOW.get(
-                `${threadId.toString()}__${connectionId.toString()}`,
-              ).catch(() => null);
-              if (existingInstance && (await existingInstance.status()).status === 'running') {
-                log('[ZERO_WORKFLOW] Thread already processing:', isProcessing, threadId);
-                continue;
-              }
-              const instance = await env.THREAD_WORKFLOW.create({
-                id: `${threadId.toString()}__${connectionId.toString()}`,
-                params: { connectionId, threadId, providerId: foundConnection.providerId },
-              });
-              log('[ZERO_WORKFLOW] Created instance:', {
-                id: instance.id,
-                status: await instance.status(),
-              });
-              log('[ZERO_WORKFLOW] Sleeping for 4 seconds:', threadId);
-              await step.sleep('[ZERO_WORKFLOW]', 4000);
-              log('[ZERO_WORKFLOW] Done sleeping:', threadId);
-              await env.gmail_processing_threads.delete(threadId.toString());
-            } catch (error) {
-              log('[ZERO_WORKFLOW] Failed to process thread:', {
-                threadId,
-                connectionId,
-                error: error instanceof Error ? error.message : String(error),
-              });
+        await step.do(
+          `[ZERO_WORKFLOW] Send Thread Workflow Instances ${connectionId}`,
+          async () => {
+            const maxConcurrentThreads = 5;
+            const delayBetweenBatches = 2000;
 
-              try {
-                await env.gmail_processing_threads.delete(threadId.toString());
-              } catch (cleanupError) {
-                log('[ZERO_WORKFLOW] Failed to cleanup processing flag:', {
-                  threadId,
-                  error:
-                    cleanupError instanceof Error ? cleanupError.message : String(cleanupError),
-                });
+            for (let i = 0; i < threadsToProcess.length; i += maxConcurrentThreads) {
+              const batch = threadsToProcess.slice(i, i + maxConcurrentThreads);
+
+              await Promise.all(
+                batch.map(async (threadId) => {
+                  try {
+                    const isProcessing = await env.gmail_processing_threads.get(
+                      threadId.toString(),
+                    );
+                    if (isProcessing) {
+                      log('[ZERO_WORKFLOW] Thread already processing:', isProcessing, threadId);
+                      return;
+                    }
+                    await env.gmail_processing_threads.put(threadId.toString(), 'true', {
+                      expirationTtl: 1800,
+                    });
+                    // const existingInstance = await env.THREAD_WORKFLOW.get(
+                    //   `${threadId.toString()}__${connectionId.toString()}`,
+                    // ).catch(() => null);
+                    // if (
+                    //   existingInstance &&
+                    //   (await existingInstance.status()).status === 'running'
+                    // ) {
+                    //   log('[ZERO_WORKFLOW] Thread already processing:', isProcessing, threadId);
+                    //   await env.gmail_processing_threads.delete(threadId.toString());
+                    //   return;
+                    // }
+                    const instance = await env.THREAD_WORKFLOW.create({
+                      //   id: `${threadId.toString()}__${connectionId.toString()}`,
+                      params: { connectionId, threadId, providerId: foundConnection.providerId },
+                    });
+                    log('[ZERO_WORKFLOW] Created instance:', {
+                      id: instance.id,
+                      status: await instance.status(),
+                    });
+                  } catch (error) {
+                    log('[ZERO_WORKFLOW] Failed to process thread:', {
+                      threadId,
+                      connectionId,
+                      error: error instanceof Error ? error.message : String(error),
+                    });
+
+                    try {
+                      await env.gmail_processing_threads.delete(threadId.toString());
+                    } catch (cleanupError) {
+                      log('[ZERO_WORKFLOW] Failed to cleanup processing flag:', {
+                        threadId,
+                        error:
+                          cleanupError instanceof Error
+                            ? cleanupError.message
+                            : String(cleanupError),
+                      });
+                    }
+                  }
+                }),
+              );
+
+              if (i + maxConcurrentThreads < threadsToProcess.length) {
+                log('[ZERO_WORKFLOW] Sleeping between batches:', delayBetweenBatches);
+                await step.sleep(
+                  `[ZERO_WORKFLOW] Sleeping between batches ${i} ${threadsToProcess.length}`,
+                  delayBetweenBatches,
+                );
               }
-              throw error;
             }
-          }
-        });
+          },
+        );
+      } else {
+        log('[ZERO_WORKFLOW] Unsupported provider:', foundConnection.providerId);
+        throw new Error(`Unsupported provider: ${foundConnection.providerId}`);
+      }
 
-        try {
-          await env.gmail_processing_threads.delete(historyProcessingKey);
-          log('[ZERO_WORKFLOW] Cleared processing flag for history:', historyProcessingKey);
-        } catch (cleanupError) {
-          log('[ZERO_WORKFLOW] Failed to clear history processing flag:', {
-            historyProcessingKey,
-            error: cleanupError instanceof Error ? cleanupError.message : String(cleanupError),
-          });
-        }
+      try {
+        await env.gmail_processing_threads.delete(historyProcessingKey);
+        log('[ZERO_WORKFLOW] Cleared processing flag for history:', historyProcessingKey);
+      } catch (cleanupError) {
+        log('[ZERO_WORKFLOW] Failed to clear history processing flag:', {
+          historyProcessingKey,
+          error: cleanupError instanceof Error ? cleanupError.message : String(cleanupError),
+        });
       }
+
       this.ctx.waitUntil(conn.end());
     } catch (error) {
       const historyProcessingKey = `history_${event.payload.connectionId}__${event.payload.historyId}`;
@@ -377,15 +440,15 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
       if (providerId === EProviders.google) {
         log('[THREAD_WORKFLOW] Processing Google provider workflow');
         const { db, conn } = createDb(env.HYPERDRIVE.connectionString);
+
         const foundConnection = await step.do(
-          `[ZERO] Find Connection ${connectionId}`,
+          `[THREAD_WORKFLOW] Find Connection ${connectionId}`,
           async () => {
             log('[THREAD_WORKFLOW] Finding connection:', connectionId);
             const [foundConnection] = await db
               .select()
               .from(connection)
               .where(eq(connection.id, connectionId.toString()));
-            this.ctx.waitUntil(conn.end());
             if (!foundConnection) throw new Error(`Connection not found ${connectionId}`);
             if (!foundConnection.accessToken || !foundConnection.refreshToken)
               throw new Error(`Connection is not authorized ${connectionId}`);
@@ -394,19 +457,28 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
           },
         );
         const driver = connectionToDriver(foundConnection);
-        const thread = await step.do(`[ZERO] Get Thread ${threadId}`, async () => {
-          log('[THREAD_WORKFLOW] Getting thread:', threadId);
-          const thread = await driver.get(threadId.toString());
-          await notifyUser({
-            connectionId: connectionId.toString(),
-            result: thread,
-            threadId: threadId.toString(),
-          });
-          log('[THREAD_WORKFLOW] Found thread with messages:', thread.messages.length);
-          return thread;
-        });
+        const thread = await step.do(
+          `[THREAD_WORKFLOW] Get Thread ${threadId} ${connectionId}`,
+          async () => {
+            log('[THREAD_WORKFLOW] Getting thread:', threadId);
+            const thread = await driver.get(threadId.toString());
+            await notifyUser({
+              connectionId: connectionId.toString(),
+              result: thread,
+              threadId: threadId.toString(),
+            });
+            log('[THREAD_WORKFLOW] Found thread with messages:', thread.messages.length);
+            return thread;
+          },
+        );
+
+        if (!thread.messages || thread.messages.length === 0) {
+          log('[THREAD_WORKFLOW] Thread has no messages, skipping processing');
+          return;
+        }
+
         const messagesToVectorize = await step.do(
-          `[ZERO] Get Thread Messages ${threadId}`,
+          `[THREAD_WORKFLOW] Get Thread Messages ${threadId} ${connectionId}`,
           async () => {
             log('[THREAD_WORKFLOW] Finding messages to vectorize');
             log('[THREAD_WORKFLOW] Getting message IDs from thread');
@@ -428,129 +500,189 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
             return messagesToVectorize;
           },
         );
-        const finalEmbeddings: VectorizeVector[] = await step.do(
-          `[ZERO] Vectorize Messages`,
-          async () => {
-            log(
-              '[THREAD_WORKFLOW] Starting message vectorization for',
-              messagesToVectorize.length,
-              'messages',
-            );
-            return await Promise.all(
-              messagesToVectorize.map(async (message) => {
-                return step.do(`[ZERO] Vectorize Message ${message.id}`, async () => {
-                  log('[THREAD_WORKFLOW] Converting message to XML:', message.id);
-                  const prompt = await messageToXML(message);
-                  if (!prompt) throw new Error(`Message has no prompt ${message.id}`);
-                  log('[THREAD_WORKFLOW] Got XML prompt for message:', message.id);
-                  log('[THREAD_WORKFLOW] Message:', message);
-
-                  const SummarizeMessagePrompt = await step.do(
-                    `[ZERO] Get Summarize Message Prompt ${message.id}`,
-                    async () => {
-                      log(
-                        '[THREAD_WORKFLOW] Getting summarize prompt for connection:',
-                        message.connectionId ?? '',
-                        message,
-                      );
-                      return await getPrompt(
-                        getPromptName(message.connectionId ?? '', EPrompts.SummarizeMessage),
-                        SummarizeMessage,
-                      );
-                    },
-                  );
-                  log('[THREAD_WORKFLOW] Got summarize prompt for message:', message.id);
-
-                  const summary: string = await step.do(
-                    `[ZERO] Summarize Message ${message.id}`,
-                    async () => {
-                      try {
-                        log('[THREAD_WORKFLOW] Generating summary for message:', message.id);
-                        const messages = [
-                          { role: 'system', content: SummarizeMessagePrompt },
-                          {
-                            role: 'user',
-                            content: prompt,
-                          },
-                        ];
-                        const response: any = await env.AI.run(
-                          '@cf/meta/llama-4-scout-17b-16e-instruct',
-                          {
-                            messages,
-                          },
-                        );
-                        log(
-                          `[THREAD_WORKFLOW] Summary generated for message ${message.id}:`,
-                          response,
-                        );
-                        return 'response' in response ? response.response : response;
-                      } catch (error) {
-                        log('[THREAD_WORKFLOW] Failed to generate summary for message:', {
-                          messageId: message.id,
-                          error: error instanceof Error ? error.message : String(error),
-                        });
-                        throw error;
-                      }
-                    },
-                  );
 
-                  const embeddingVector = await step.do(
-                    `[ZERO] Get Message Embedding Vector ${message.id}`,
-                    async () => {
-                      try {
-                        log('[THREAD_WORKFLOW] Getting embedding vector for message:', message.id);
-                        const embeddingVector = await getEmbeddingVector(summary);
-                        log('[THREAD_WORKFLOW] Got embedding vector for message:', message.id);
-                        return embeddingVector;
-                      } catch (error) {
-                        log('[THREAD_WORKFLOW] Failed to get embedding vector for message:', {
-                          messageId: message.id,
-                          error: error instanceof Error ? error.message : String(error),
-                        });
-                        throw error;
-                      }
-                    },
-                  );
+        if (messagesToVectorize.length === 0) {
+          log('[THREAD_WORKFLOW] No messages to vectorize, skipping vectorization');
+        } else {
+          const finalEmbeddings: VectorizeVector[] = await step.do(
+            `[THREAD_WORKFLOW] Vectorize Messages ${threadId} ${connectionId}`,
+            async () => {
+              log(
+                '[THREAD_WORKFLOW] Starting message vectorization for',
+                messagesToVectorize.length,
+                'messages',
+              );
 
-                  if (!embeddingVector)
-                    throw new Error(`Message Embedding vector is null ${message.id}`);
-
-                  return {
-                    id: message.id,
-                    metadata: {
-                      connection: message.connectionId ?? '',
-                      thread: message.threadId ?? '',
-                      summary,
-                    },
-                    values: embeddingVector,
-                  } satisfies VectorizeVector;
-                });
-              }),
-            );
-          },
-        );
-        log('[THREAD_WORKFLOW] Generated embeddings for all messages');
+              const maxConcurrentMessages = 3;
+              const results: VectorizeVector[] = [];
 
-        await step.do(
-          `[ZERO] Thread Messages Vectors ${threadId} / ${finalEmbeddings.length}`,
-          async () => {
-            try {
-              log('[THREAD_WORKFLOW] Upserting message vectors:', finalEmbeddings.length);
-              await env.VECTORIZE_MESSAGE.upsert(finalEmbeddings);
-              log('[THREAD_WORKFLOW] Successfully upserted message vectors');
-            } catch (error) {
-              log('[THREAD_WORKFLOW] Failed to upsert message vectors:', {
-                threadId,
-                vectorCount: finalEmbeddings.length,
-                error: error instanceof Error ? error.message : String(error),
-              });
-              throw error;
-            }
-          },
-        );
+              for (let i = 0; i < messagesToVectorize.length; i += maxConcurrentMessages) {
+                const batch = messagesToVectorize.slice(i, i + maxConcurrentMessages);
+                const batchResults = await Promise.all(
+                  batch.map(async (message) => {
+                    return step.do(
+                      `[THREAD_WORKFLOW] Vectorize Message ${message.id} ${threadId}`,
+                      async () => {
+                        try {
+                          log('[THREAD_WORKFLOW] Converting message to XML:', message.id);
+                          const prompt = await messageToXML(message);
+                          if (!prompt) {
+                            log('[THREAD_WORKFLOW] Message has no prompt, skipping:', message.id);
+                            return null;
+                          }
+                          log('[THREAD_WORKFLOW] Got XML prompt for message:', message.id);
+
+                          const SummarizeMessagePrompt = await step.do(
+                            `[THREAD_WORKFLOW] Get Summarize Message Prompt ${message.id} ${threadId}`,
+                            async () => {
+                              log(
+                                '[THREAD_WORKFLOW] Getting summarize prompt for connection:',
+                                message.connectionId ?? '',
+                              );
+                              return await getPrompt(
+                                getPromptName(
+                                  message.connectionId ?? '',
+                                  EPrompts.SummarizeMessage,
+                                ),
+                                SummarizeMessage,
+                              );
+                            },
+                          );
+                          log('[THREAD_WORKFLOW] Got summarize prompt for message:', message.id);
+
+                          const summary: string = await step.do(
+                            `[THREAD_WORKFLOW] Summarize Message ${message.id} ${threadId}`,
+                            async () => {
+                              try {
+                                log(
+                                  '[THREAD_WORKFLOW] Generating summary for message:',
+                                  message.id,
+                                );
+                                const messages = [
+                                  { role: 'system', content: SummarizeMessagePrompt },
+                                  {
+                                    role: 'user',
+                                    content: prompt,
+                                  },
+                                ];
+                                const response: any = await env.AI.run(
+                                  '@cf/meta/llama-4-scout-17b-16e-instruct',
+                                  {
+                                    messages,
+                                  },
+                                );
+                                log(
+                                  `[THREAD_WORKFLOW] Summary generated for message ${message.id}:`,
+                                  response,
+                                );
+                                const summary =
+                                  'response' in response ? response.response : response;
+                                if (!summary || typeof summary !== 'string') {
+                                  throw new Error(
+                                    `Invalid summary response for message ${message.id}`,
+                                  );
+                                }
+                                return summary;
+                              } catch (error) {
+                                log('[THREAD_WORKFLOW] Failed to generate summary for message:', {
+                                  messageId: message.id,
+                                  error: error instanceof Error ? error.message : String(error),
+                                });
+                                throw error;
+                              }
+                            },
+                          );
+
+                          const embeddingVector = await step.do(
+                            `[THREAD_WORKFLOW] Get Message Embedding Vector ${message.id} ${threadId}`,
+                            async () => {
+                              try {
+                                log(
+                                  '[THREAD_WORKFLOW] Getting embedding vector for message:',
+                                  message.id,
+                                );
+                                const embeddingVector = await getEmbeddingVector(summary);
+                                log(
+                                  '[THREAD_WORKFLOW] Got embedding vector for message:',
+                                  message.id,
+                                );
+                                return embeddingVector;
+                              } catch (error) {
+                                log(
+                                  '[THREAD_WORKFLOW] Failed to get embedding vector for message:',
+                                  {
+                                    messageId: message.id,
+                                    error: error instanceof Error ? error.message : String(error),
+                                  },
+                                );
+                                throw error;
+                              }
+                            },
+                          );
+
+                          if (!embeddingVector)
+                            throw new Error(`Message Embedding vector is null ${message.id}`);
+
+                          return {
+                            id: message.id,
+                            metadata: {
+                              connection: message.connectionId ?? '',
+                              thread: message.threadId ?? '',
+                              summary,
+                            },
+                            values: embeddingVector,
+                          } satisfies VectorizeVector;
+                        } catch (error) {
+                          log('[THREAD_WORKFLOW] Failed to vectorize message:', {
+                            messageId: message.id,
+                            error: error instanceof Error ? error.message : String(error),
+                          });
+                          return null;
+                        }
+                      },
+                    );
+                  }),
+                );
+
+                const validResults = batchResults.filter(
+                  (result): result is NonNullable<typeof result> => result !== null,
+                );
+                results.push(...validResults);
+
+                if (i + maxConcurrentMessages < messagesToVectorize.length) {
+                  log('[THREAD_WORKFLOW] Sleeping between message batches');
+                  await step.sleep('[THREAD_WORKFLOW]', 1000);
+                }
+              }
+
+              return results;
+            },
+          );
+          log('[THREAD_WORKFLOW] Generated embeddings for all messages');
+
+          if (finalEmbeddings.length > 0) {
+            await step.do(
+              `[THREAD_WORKFLOW] Thread Messages Vectors ${threadId} ${connectionId} ${finalEmbeddings.length}`,
+              async () => {
+                try {
+                  log('[THREAD_WORKFLOW] Upserting message vectors:', finalEmbeddings.length);
+                  await env.VECTORIZE_MESSAGE.upsert(finalEmbeddings);
+                  log('[THREAD_WORKFLOW] Successfully upserted message vectors');
+                } catch (error) {
+                  log('[THREAD_WORKFLOW] Failed to upsert message vectors:', {
+                    threadId,
+                    vectorCount: finalEmbeddings.length,
+                    error: error instanceof Error ? error.message : String(error),
+                  });
+                  throw error;
+                }
+              },
+            );
+          }
+        }
 
         const existingThreadSummary = await step.do(
-          `[ZERO] Get Thread Summary ${threadId}`,
+          `[THREAD_WORKFLOW] Get Thread Summary ${threadId} ${connectionId}`,
           async () => {
             log('[THREAD_WORKFLOW] Getting existing thread summary for:', threadId);
             const threadSummary = await env.VECTORIZE.getByIds([threadId.toString()]);
@@ -563,23 +695,26 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
           },
         );
 
-        const finalSummary = await step.do(`[ZERO] Get Final Summary ${threadId}`, async () => {
-          log('[THREAD_WORKFLOW] Generating final thread summary');
-          if (existingThreadSummary) {
-            log('[THREAD_WORKFLOW] Using existing summary as context');
-            return await summarizeThread(
-              connectionId.toString(),
-              thread.messages,
-              existingThreadSummary.summary,
-            );
-          } else {
-            log('[THREAD_WORKFLOW] Generating new summary without context');
-            return await summarizeThread(connectionId.toString(), thread.messages);
-          }
-        });
+        const finalSummary = await step.do(
+          `[THREAD_WORKFLOW] Get Final Summary ${threadId} ${connectionId}`,
+          async () => {
+            log('[THREAD_WORKFLOW] Generating final thread summary');
+            if (existingThreadSummary) {
+              log('[THREAD_WORKFLOW] Using existing summary as context');
+              return await summarizeThread(
+                connectionId.toString(),
+                thread.messages,
+                existingThreadSummary.summary,
+              );
+            } else {
+              log('[THREAD_WORKFLOW] Generating new summary without context');
+              return await summarizeThread(connectionId.toString(), thread.messages);
+            }
+          },
+        );
 
         const userAccountLabels = await step.do(
-          `[ZERO] Get user-account labels ${connectionId}`,
+          `[THREAD_WORKFLOW] Get user-account labels ${connectionId}`,
           async () => {
             try {
               const userAccountLabels = await driver.getUserLabels();
@@ -597,7 +732,7 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
         if (finalSummary) {
           log('[THREAD_WORKFLOW] Got final summary, processing labels');
           const userLabels = await step.do(
-            `[ZERO] Get user-defined labels ${connectionId}`,
+            `[THREAD_WORKFLOW] Get user-defined labels ${connectionId}`,
             async () => {
               log('[THREAD_WORKFLOW] Getting user labels for connection:', connectionId);
               let userLabels: { name: string; usecase: string }[] = [];
@@ -606,7 +741,16 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
                 try {
                   log('[THREAD_WORKFLOW] Parsing existing connection labels');
                   const parsed = JSON.parse(connectionLabels);
-                  userLabels = parsed;
+                  if (
+                    Array.isArray(parsed) &&
+                    parsed.every(
+                      (label) => typeof label === 'object' && label.name && label.usecase,
+                    )
+                  ) {
+                    userLabels = parsed;
+                  } else {
+                    throw new Error('Invalid label format');
+                  }
                 } catch {
                   log('[THREAD_WORKFLOW] Failed to parse labels, using defaults');
                   await env.connection_labels.put(
@@ -628,7 +772,7 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
           );
 
           const generatedLabels = await step.do(
-            `[ZERO] Generate Thread Labels ${threadId} / ${thread.messages.length}`,
+            `[THREAD_WORKFLOW] Generate Thread Labels ${threadId} ${connectionId} ${thread.messages.length}`,
             async () => {
               try {
                 log('[THREAD_WORKFLOW] Generating labels for thread:', threadId);
@@ -645,42 +789,48 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
                   log('[THREAD_WORKFLOW] Labels generated:', labelsResponse.response);
                   const labels: string[] = labelsResponse?.response
                     ?.split(',')
+                    .map((e: string) => e.trim())
+                    .filter((e: string) => e.length > 0)
                     .filter((e: string) =>
                       userLabels.find((label) => label.name.toLowerCase() === e.toLowerCase()),
                     );
                   return labels;
                 } else {
                   log('[THREAD_WORKFLOW] No labels generated');
+                  return [];
                 }
               } catch (error) {
                 log('[THREAD_WORKFLOW] Failed to generate labels for thread:', {
                   threadId,
                   error: error instanceof Error ? error.message : String(error),
                 });
-                throw error;
+                return [];
               }
             },
           );
 
-          if (generatedLabels) {
-            await step.do(`[ZERO] Modify Thread Labels ${threadId}`, async () => {
-              log('[THREAD_WORKFLOW] Modifying thread labels:', generatedLabels);
-              const validLabelIds = generatedLabels
-                .map((name) => userAccountLabels.find((e) => e.name === name)?.id)
-                .filter((id): id is string => id !== undefined && id !== '');
-
-              if (validLabelIds.length > 0) {
-                await driver.modifyLabels([threadId.toString()], {
-                  addLabels: validLabelIds,
-                  removeLabels: [],
-                });
-              }
-              log('[THREAD_WORKFLOW] Successfully modified thread labels');
-            });
+          if (generatedLabels && generatedLabels.length > 0) {
+            await step.do(
+              `[THREAD_WORKFLOW] Modify Thread Labels ${threadId} ${connectionId}`,
+              async () => {
+                log('[THREAD_WORKFLOW] Modifying thread labels:', generatedLabels);
+                const validLabelIds = generatedLabels
+                  .map((name) => userAccountLabels.find((e) => e.name === name)?.id)
+                  .filter((id): id is string => id !== undefined && id !== '');
+
+                if (validLabelIds.length > 0) {
+                  await driver.modifyLabels([threadId.toString()], {
+                    addLabels: validLabelIds,
+                    removeLabels: [],
+                  });
+                }
+                log('[THREAD_WORKFLOW] Successfully modified thread labels');
+              },
+            );
           }
 
           const embeddingVector = await step.do(
-            `[ZERO] Get Thread Embedding Vector ${threadId}`,
+            `[THREAD_WORKFLOW] Get Thread Embedding Vector ${threadId} ${connectionId}`,
             async () => {
               log('[THREAD_WORKFLOW] Getting thread embedding vector');
               const embeddingVector = await getEmbeddingVector(finalSummary);
@@ -689,7 +839,10 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
             },
           );
 
-          if (!embeddingVector) return console.error('Thread Embedding vector is null');
+          if (!embeddingVector) {
+            log('[THREAD_WORKFLOW] Thread Embedding vector is null, skipping vector upsert');
+            return;
+          }
 
           try {
             log('[THREAD_WORKFLOW] Upserting thread vector');
@@ -720,6 +873,11 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
             thread.messages.length,
           );
         }
+
+        this.ctx.waitUntil(conn.end());
+      } else {
+        log('[THREAD_WORKFLOW] Unsupported provider:', providerId);
+        throw new Error(`Unsupported provider: ${providerId}`);
       }
     } catch (error) {
       log('[THREAD_WORKFLOW] Error in workflow:', error);
@@ -737,6 +895,9 @@ export class ThreadWorkflow extends WorkflowEntrypoint<Env, Params> {
 
 export async function htmlToText(decodedBody: string): Promise<string> {
   try {
+    if (!decodedBody || typeof decodedBody !== 'string') {
+      return '';
+    }
     const $ = cheerio.load(decodedBody);
     $('script').remove();
     $('style').remove();
@@ -747,27 +908,49 @@ export async function htmlToText(decodedBody: string): Promise<string> {
       .trim();
   } catch (error) {
     log('Error extracting text from HTML:', error);
-    throw new Error('Failed to extract text from HTML');
+    return '';
   }
 }
 
+const escapeXml = (text: string): string => {
+  if (!text) return '';
+  return text
+    .replace(/&/g, '&amp;')
+    .replace(/</g, '&lt;')
+    .replace(/>/g, '&gt;')
+    .replace(/"/g, '&quot;')
+    .replace(/'/g, '&apos;');
+};
+
 const messageToXML = async (message: ParsedMessage) => {
   try {
     if (!message.decodedBody) return null;
     const body = await htmlToText(message.decodedBody || '');
     log('[MESSAGE_TO_XML] Body', body);
-    if (!body || (body?.length || 20) < 20) {
-      log('Skipping message with body length < 20', body);
+    if (!body || body.length < 10) {
+      log('Skipping message with body length < 10', body);
       return null;
     }
+
+    const safeSenderName = escapeXml(message.sender?.name || 'Unknown');
+    const safeSubject = escapeXml(message.subject || '');
+    const safeDate = escapeXml(message.receivedOn || '');
+
+    const toElements = (message.to || [])
+      .map((r) => `<to>${escapeXml(r?.email || '')}</to>`)
+      .join('');
+    const ccElements = (message.cc || [])
+      .map((r) => `<cc>${escapeXml(r?.email || '')}</cc>`)
+      .join('');
+
     return `
         <message>
-          <from>${message.sender.name}</from>
-          ${message.to.map((r) => `<to>${r.email}</to>`).join('')}
-          ${message.cc ? message.cc.map((r) => `<cc>${r.email}</cc>`).join('') : ''}
-          <date>${message.receivedOn}</date>
-          <subject>${message.subject}</subject>
-          <body>${body}</body>
+          <from>${safeSenderName}</from>
+          ${toElements}
+          ${ccElements}
+          <date>${safeDate}</date>
+          <subject>${safeSubject}</subject>
+          <body>${escapeXml(body)}</body>
         </message>
         `;
   } catch (error) {
@@ -775,7 +958,7 @@ const messageToXML = async (message: ParsedMessage) => {
       messageId: message.id,
       error: error instanceof Error ? error.message : String(error),
     });
-    throw error;
+    return null;
   }
 };
 
@@ -785,6 +968,11 @@ export const getPromptName = (connectionId: string, prompt: EPrompts) => {
 
 export const getPrompt = async (promptName: string, fallback: string) => {
   try {
+    if (!promptName || typeof promptName !== 'string') {
+      log('[GET_PROMPT] Invalid prompt name:', promptName);
+      return fallback;
+    }
+
     const existingPrompt = await env.prompts_storage.get(promptName);
     if (!existingPrompt) {
       await env.prompts_storage.put(promptName, fallback);
@@ -802,16 +990,21 @@ export const getPrompt = async (promptName: string, fallback: string) => {
 
 export const getEmbeddingVector = async (text: string) => {
   try {
+    if (!text || typeof text !== 'string' || text.trim().length === 0) {
+      log('[getEmbeddingVector] Empty or invalid text provided');
+      return null;
+    }
+
     const embeddingResponse = await env.AI.run(
       '@cf/baai/bge-large-en-v1.5',
-      { text },
+      { text: text.trim() },
       {
         gateway: {
           id: 'vectorize-save',
         },
       },
     );
-    const embeddingVector = embeddingResponse.data[0];
+    const embeddingVector = (embeddingResponse as any).data?.[0];
     return embeddingVector ?? null;
   } catch (error) {
     log('[getEmbeddingVector] failed', error);
@@ -820,57 +1013,82 @@ export const getEmbeddingVector = async (text: string) => {
 };
 
 const isValidUUID = (string: string) => {
+  if (!string || typeof string !== 'string') return false;
   return z.string().uuid().safeParse(string).success;
 };
 
 const getParticipants = (messages: ParsedMessage[]) => {
+  if (!messages || !Array.isArray(messages) || messages.length === 0) {
+    return [];
+  }
+
   const result = new Map<Sender['email'], Sender['name'] | ''>();
   const setIfUnset = (sender: Sender) => {
-    if (!result.has(sender.email)) result.set(sender.email, sender.name);
+    if (sender?.email && !result.has(sender.email)) {
+      result.set(sender.email, sender.name || '');
+    }
   };
+
   for (const msg of messages) {
-    setIfUnset(msg.sender);
-    for (const ccParticipant of msg.cc ?? []) {
-      setIfUnset(ccParticipant);
+    if (msg?.sender) {
+      setIfUnset(msg.sender);
+    }
+    if (msg?.cc && Array.isArray(msg.cc)) {
+      for (const ccParticipant of msg.cc) {
+        if (ccParticipant) setIfUnset(ccParticipant);
+      }
     }
-    for (const toParticipant of msg.to) {
-      setIfUnset(toParticipant);
+    if (msg?.to && Array.isArray(msg.to)) {
+      for (const toParticipant of msg.to) {
+        if (toParticipant) setIfUnset(toParticipant);
+      }
     }
   }
   return Array.from(result.entries());
 };
 
 const threadToXML = async (messages: ParsedMessage[], existingSummary?: string) => {
-  const { subject, title } = messages[0];
+  if (!messages || !Array.isArray(messages) || messages.length === 0) {
+    throw new Error('No messages provided for thread XML generation');
+  }
+
+  const firstMessage = messages[0];
+  if (!firstMessage) {
+    throw new Error('First message is null or undefined');
+  }
+
+  const { subject = '', title = '' } = firstMessage;
   const participants = getParticipants(messages);
   const messagesXML = await Promise.all(messages.map(messageToXML));
+  const validMessagesXML = messagesXML.filter((xml): xml is string => xml !== null);
+
   if (existingSummary) {
     return `<thread>
-            <title>${title}</title>
-            <subject>${subject}</subject>
+            <title>${escapeXml(title)}</title>
+            <subject>${escapeXml(subject)}</subject>
             <participants>
               ${participants.map(([email, name]) => {
-                return `<participant>${name ?? email} ${name ? `< ${email} >` : ''}</participant>`;
+                return `<participant>${escapeXml(name || email)} ${name ? `< ${escapeXml(email)} >` : ''}</participant>`;
               })}
             </participants>
             <existing_summary>
-              ${existingSummary}
+              ${escapeXml(existingSummary)}
             </existing_summary>
             <new_messages>
-                ${messagesXML.map((e) => e + '\n')}
+                ${validMessagesXML.map((e) => e + '\n')}
             </new_messages>
         </thread>`;
   }
   return `<thread>
-          <title>${title}</title>
-          <subject>${subject}</subject>
+          <title>${escapeXml(title)}</title>
+          <subject>${escapeXml(subject)}</subject>
           <participants>
             ${participants.map(([email, name]) => {
-              return `<participant>${name} < ${email} ></participant>`;
+              return `<participant>${escapeXml(name || email)} < ${escapeXml(email)} ></participant>`;
             })}
           </participants>
           <messages>
-              ${messagesXML.map((e) => e + '\n')}
+              ${validMessagesXML.map((e) => e + '\n')}
           </messages>
       </thread>`;
 };
@@ -881,8 +1099,23 @@ const summarizeThread = async (
   existingSummary?: string,
 ): Promise<string | null> => {
   try {
+    if (!messages || !Array.isArray(messages) || messages.length === 0) {
+      log('[SUMMARIZE_THREAD] No messages provided for summarization');
+      return null;
+    }
+
+    if (!connectionId || typeof connectionId !== 'string') {
+      log('[SUMMARIZE_THREAD] Invalid connection ID provided');
+      return null;
+    }
+
+    const prompt = await threadToXML(messages, existingSummary);
+    if (!prompt) {
+      log('[SUMMARIZE_THREAD] Failed to generate thread XML');
+      return null;
+    }
+
     if (existingSummary) {
-      const prompt = await threadToXML(messages, existingSummary);
       const ReSummarizeThreadPrompt = await getPrompt(
         getPromptName(connectionId, EPrompts.ReSummarizeThread),
         ReSummarizeThread,
@@ -897,9 +1130,9 @@ const summarizeThread = async (
       const response: any = await env.AI.run('@cf/meta/llama-3.3-70b-instruct-fp8-fast', {
         messages: promptMessages,
       });
-      return response.response ?? null;
+      const summary = response?.response;
+      return typeof summary === 'string' ? summary : null;
     } else {
-      const prompt = await threadToXML(messages, existingSummary);
       const SummarizeThreadPrompt = await getPrompt(
         getPromptName(connectionId, EPrompts.SummarizeThread),
         SummarizeThread,
@@ -914,12 +1147,13 @@ const summarizeThread = async (
       const response: any = await env.AI.run('@cf/meta/llama-3.3-70b-instruct-fp8-fast', {
         messages: promptMessages,
       });
-      return response.response ?? null;
+      const summary = response?.response;
+      return typeof summary === 'string' ? summary : null;
     }
   } catch (error) {
     log('[SUMMARIZE_THREAD] Failed to summarize thread:', {
       connectionId,
-      messageCount: messages.length,
+      messageCount: messages?.length || 0,
       hasExistingSummary: !!existingSummary,
       error: error instanceof Error ? error.message : String(error),
     });
diff --git a/apps/server/src/trpc/routes/mail.ts b/apps/server/src/trpc/routes/mail.ts
index 623ffcb2db..ec9a6d72a4 100644
--- a/apps/server/src/trpc/routes/mail.ts
+++ b/apps/server/src/trpc/routes/mail.ts
@@ -1,11 +1,18 @@
-import { activeDriverProcedure, createRateLimiterMiddleware, router } from '../trpc';
+import {
+  activeDriverProcedure,
+  createRateLimiterMiddleware,
+  router,
+  privateProcedure,
+} from '../trpc';
 import { updateWritingStyleMatrix } from '../../services/writing-style-service';
 import { deserializeFiles, serializedFileSchema } from '../../lib/schemas';
 import { defaultPageSize, FOLDERS, LABELS } from '../../lib/utils';
 import { IGetThreadResponseSchema } from '../../lib/driver/types';
+import { processEmailHtml } from '../../lib/email-processor';
 import type { DeleteAllSpamResponse } from '../../types';
 import { getZeroAgent } from '../../lib/server-utils';
 import { env } from 'cloudflare:workers';
+import { TRPCError } from '@trpc/server';
 import { z } from 'zod';
 
 const senderSchema = z.object({
@@ -395,4 +402,32 @@ export const mailRouter = router({
     const agent = await getZeroAgent(activeConnection.id);
     return agent.getEmailAliases();
   }),
+  processEmailContent: privateProcedure
+    .input(
+      z.object({
+        html: z.string(),
+        shouldLoadImages: z.boolean(),
+        theme: z.enum(['light', 'dark']),
+      }),
+    )
+    .mutation(async ({ input }) => {
+      try {
+        const { processedHtml, hasBlockedImages } = processEmailHtml({
+          html: input.html,
+          shouldLoadImages: input.shouldLoadImages,
+          theme: input.theme,
+        });
+
+        return {
+          processedHtml,
+          hasBlockedImages,
+        };
+      } catch (error) {
+        console.error('Error processing email content:', error);
+        throw new TRPCError({
+          code: 'INTERNAL_SERVER_ERROR',
+          message: 'Failed to process email content',
+        });
+      }
+    }),
 });
diff --git a/apps/server/wrangler.jsonc b/apps/server/wrangler.jsonc
index 87fbeda9b2..5071e05f64 100644
--- a/apps/server/wrangler.jsonc
+++ b/apps/server/wrangler.jsonc
@@ -81,17 +81,17 @@
       ],
       "workflows": [
         {
-          "name": "main-workflow",
+          "name": "main-workflow-staging",
           "binding": "MAIN_WORKFLOW",
           "class_name": "MainWorkflow",
         },
         {
-          "name": "zero-workflow",
+          "name": "zero-workflow-staging",
           "binding": "ZERO_WORKFLOW",
           "class_name": "ZeroWorkflow",
         },
         {
-          "name": "thread-workflow",
+          "name": "thread-workflow-staging",
           "binding": "THREAD_WORKFLOW",
           "class_name": "ThreadWorkflow",
         },
@@ -229,17 +229,17 @@
       ],
       "workflows": [
         {
-          "name": "main-workflow",
+          "name": "main-workflow-staging",
           "binding": "MAIN_WORKFLOW",
           "class_name": "MainWorkflow",
         },
         {
-          "name": "zero-workflow",
+          "name": "zero-workflow-staging",
           "binding": "ZERO_WORKFLOW",
           "class_name": "ZeroWorkflow",
         },
         {
-          "name": "thread-workflow",
+          "name": "thread-workflow-staging",
           "binding": "THREAD_WORKFLOW",
           "class_name": "ThreadWorkflow",
         },
@@ -316,17 +316,17 @@
       ],
       "workflows": [
         {
-          "name": "main-workflow",
+          "name": "main-workflow-prod",
           "binding": "MAIN_WORKFLOW",
           "class_name": "MainWorkflow",
         },
         {
-          "name": "zero-workflow",
+          "name": "zero-workflow-prod",
           "binding": "ZERO_WORKFLOW",
           "class_name": "ZeroWorkflow",
         },
         {
-          "name": "thread-workflow",
+          "name": "thread-workflow-prod",
           "binding": "THREAD_WORKFLOW",
           "class_name": "ThreadWorkflow",
         },