Onboarding process emails

apps/web/app/(app)/[emailAccountId]/onboarding/page.tsx

@@ -6,6 +6,8 @@ import { OnboardingContent } from "@/app/(app)/[emailAccountId]/onboarding/Onboa
 import { fetchUserAndStoreUtms } from "@/app/(landing)/welcome/utms";
 import { auth } from "@/utils/auth";
 
+export const maxDuration = 300;
+
 export const metadata: Metadata = {
   title: "Onboarding | Inbox Zero",
   description: "Learn how Inbox Zero works and get set up.",

apps/web/utils/actions/rule.ts

@@ -1,6 +1,7 @@
 "use server";
 
 import { revalidatePath } from "next/cache";
+import { after } from "next/server";
 import {
   createRuleBody,
   updateRuleBody,
@@ -39,6 +40,8 @@ import { resolveLabelNameAndId } from "@/utils/label/resolve-label";
 import type { Logger } from "@/utils/logger";
 import { validateGmailLabelName } from "@/utils/gmail/label-validation";
 import { isGoogleProvider } from "@/utils/email/provider-types";
+import { bulkProcessInboxEmails } from "@/utils/ai/choose-rule/bulk-process-emails";
+import { getEmailAccountWithAi } from "@/utils/user/get";
 
 export const createRuleAction = actionClient
   .metadata({ name: "createRule" })
@@ -272,10 +275,7 @@ export const createRulesOnboardingAction = actionClient
         }
       }
 
-      const emailAccount = await prisma.emailAccount.findUnique({
-        where: { id: emailAccountId },
-        select: { rulesPrompt: true },
-      });
+      const emailAccount = await getEmailAccountWithAi({ emailAccountId });
       if (!emailAccount) throw new SafeError("User not found");
 
       const promises: Promise<unknown>[] = [];
@@ -410,6 +410,16 @@ export const createRulesOnboardingAction = actionClient
       }
 
       await Promise.allSettled(promises);
+
+      after(() =>
+        bulkProcessInboxEmails({
+          emailAccount,
+          provider,
+          maxEmails: 20,
+          skipArchive: true,
+          logger,
+        }),
+      );
     },
   );
 

apps/web/utils/ai/choose-rule/bulk-process-emails.ts

@@ -0,0 +1,114 @@
+import prisma from "@/utils/prisma";
+import { createEmailProvider } from "@/utils/email/provider";
+import { runRules } from "@/utils/ai/choose-rule/run-rules";
+import type { Logger } from "@/utils/logger";
+import type { EmailAccountWithAI } from "@/utils/llms/types";
+import type { ParsedMessage } from "@/utils/types";
+
+export async function bulkProcessInboxEmails({
+  emailAccount,
+  provider,
+  maxEmails,
+  skipArchive,
+  logger: log,
+}: {
+  emailAccount: EmailAccountWithAI;
+  provider: string;
+  maxEmails: number;
+  skipArchive: boolean;
+  logger: Logger;
+}) {
+  const logger = log.with({ module: "bulk-process-emails" });
+
+  logger.info("Starting bulk inbox email processing");
+
+  try {
+    const emailProvider = await createEmailProvider({
+      emailAccountId: emailAccount.id,
+      provider,
+      logger,
+    });
+
+    const [{ messages }, rules] = await Promise.all([
+      emailProvider.getMessagesByFields({
+        type: "inbox",
+        maxResults: maxEmails,
+      }),
+      prisma.rule.findMany({
+        where: {
+          emailAccountId: emailAccount.id,
+          enabled: true,
+        },
+        include: { actions: true },
+      }),
+    ]);
+
+    if (messages.length === 0) {
+      logger.info("No inbox emails to process");
+      return;
+    }
+
+    if (rules.length === 0) {
+      logger.info("No rules found");
+      return;
+    }
+
+    const uniqueMessages = getLatestMessagePerThread(messages);
+
+    logger.info("Processing emails with rules", {
+      ruleCount: rules.length,
+      emailCount: uniqueMessages.length,
+      totalFetched: messages.length,
+    });
+
+    let processedCount = 0;
+    let errorCount = 0;
+
+    for (const message of uniqueMessages) {
+      try {
+        await runRules({
+          provider: emailProvider,
+          message,
+          rules,
+          emailAccount,
+          isTest: false,
+          modelType: "economy",
+          logger,
+          skipArchive,
+        });
+        processedCount++;
+      } catch (error) {
+        errorCount++;
+        logger.error("Error processing email", {
+          messageId: message.id,
+          error,
+        });
+        // Continue processing other emails even if one fails
+      }
+    }
+
+    logger.info("Completed bulk email processing", {
+      processedCount,
+      errorCount,
+      totalEmails: uniqueMessages.length,
+    });
+  } catch (error) {
+    logger.error("Failed to process emails", { error });
+  }
+}
+
+function getLatestMessagePerThread(messages: ParsedMessage[]): ParsedMessage[] {
+  const latestByThread = new Map<string, ParsedMessage>();
+
+  for (const message of messages) {
+    const existing = latestByThread.get(message.threadId);
+    if (
+      !existing ||
+      new Date(message.date || 0) > new Date(existing.date || 0)
+    ) {
+      latestByThread.set(message.threadId, message);
+    }
+  }
+
+  return Array.from(latestByThread.values());
+}

apps/web/utils/ai/choose-rule/run-rules.ts

@@ -72,6 +72,7 @@ export async function runRules({
   isTest,
   modelType,
   logger,
+  skipArchive,
 }: {
   provider: EmailProvider;
   message: ParsedMessage;
@@ -80,6 +81,7 @@ export async function runRules({
   isTest: boolean;
   modelType: ModelType;
   logger: Logger;
+  skipArchive?: boolean;
 }): Promise<RunRulesResult[]> {
   const batchTimestamp = new Date(); // Single timestamp for this batch execution
   const { regularRules, conversationRules } = prepareRulesWithMetaRule(rules);
@@ -187,6 +189,7 @@ export async function runRules({
       modelType,
       batchTimestamp,
       logger,
+      skipArchive,
     );
 
     executedRules.push({
@@ -253,8 +256,9 @@ async function executeMatchedRule(
   modelType: ModelType,
   batchTimestamp: Date,
   logger: Logger,
+  skipArchive?: boolean,
 ) {
-  const actionItems = await getActionItemsWithAiArgs({
+  let actionItems = await getActionItemsWithAiArgs({
     message,
     emailAccount,
     selectedRule: rule,
@@ -264,6 +268,12 @@ async function executeMatchedRule(
     isTest,
   });
 
+  if (skipArchive) {
+    actionItems = actionItems.filter(
+      (item) => item.type !== ActionType.ARCHIVE,
+    );
+  }
+
   const { immediateActions, delayedActions } = groupBy(actionItems, (item) =>
     item.delayInMinutes != null && item.delayInMinutes > 0
       ? "delayedActions"

version.txt

@@ -1 +1 @@
-v2.21.2
+v2.21.3