diff --git a/apps/web/app/(landing)/welcome-upgrade/page.tsx b/apps/web/app/(landing)/welcome-upgrade/page.tsx
index d73d7d79ed..c9c7bc1a24 100644
--- a/apps/web/app/(landing)/welcome-upgrade/page.tsx
+++ b/apps/web/app/(landing)/welcome-upgrade/page.tsx
@@ -18,7 +18,7 @@ export default function WelcomeUpgradePage() {
Spend 50% less time on email
- Join 8,000+ users that use Inbox Zero
+ Join 9,000+ users that use Inbox Zero
to be more productive!
diff --git a/apps/web/app/api/ai/cold-email/controller.ts b/apps/web/app/api/ai/cold-email/controller.ts
index 9d25e5a3e8..de3d22b442 100644
--- a/apps/web/app/api/ai/cold-email/controller.ts
+++ b/apps/web/app/api/ai/cold-email/controller.ts
@@ -50,14 +50,14 @@ export async function isColdEmail({
});
if (coldEmail) {
- logger.log(`Already marked as cold email. ${email.from}`);
+ logger.info(`Already marked as cold email. ${email.from}`);
return { isColdEmail: true, reason: "ai-already-labeled" };
}
// otherwise run through ai to see if it's a cold email
const res = await aiIsColdEmail(email, user);
- logger.log(`AI is cold email: ${res.coldEmail}`);
+ logger.info(`AI is cold email: ${res.coldEmail}`);
return {
isColdEmail: !!res.coldEmail,
diff --git a/apps/web/app/api/google/webhook/process-history.ts b/apps/web/app/api/google/webhook/process-history.ts
index d03b84b8f7..05768d5469 100644
--- a/apps/web/app/api/google/webhook/process-history.ts
+++ b/apps/web/app/api/google/webhook/process-history.ts
@@ -22,7 +22,7 @@ import { runColdEmailBlocker } from "@/app/api/ai/cold-email/controller";
import { captureException } from "@/utils/error";
import { runRulesOnMessage } from "@/utils/ai/choose-rule/run-rules";
import { blockUnsubscribedEmails } from "@/app/api/google/webhook/block-unsubscribed-emails";
-import { categorizeSender } from "@/utils/actions/categorize";
+import { categorizeSender } from "@/utils/categorize/senders/categorize";
import { unwatchEmails } from "@/app/api/google/watch/controller";
import { createScopedLogger } from "@/utils/logger";
import { markMessageAsProcessing } from "@/utils/redis/message-processing";
@@ -83,7 +83,7 @@ export async function processHistoryForUser(
: undefined;
if (!premium) {
- logger.log(`Account not premium. email: ${email}`);
+ logger.info(`Account not premium. email: ${email}`);
await unwatchEmails(account);
return NextResponse.json({ ok: true });
}
@@ -147,7 +147,7 @@ export async function processHistoryForUser(
historyId - 500, // avoid going too far back
).toString();
- logger.log(
+ logger.info(
`Listing history... Start: ${startHistoryId} lastSyncedHistoryId: ${account.user.lastSyncedHistoryId} gmailHistoryId: ${startHistoryId} email: ${email}`,
);
@@ -162,7 +162,7 @@ export async function processHistoryForUser(
});
if (history.data.history) {
- logger.log(
+ logger.info(
`Processing... email: ${email} startHistoryId: ${startHistoryId} historyId: ${history.data.historyId}`,
);
@@ -188,7 +188,7 @@ export async function processHistoryForUser(
},
});
} else {
- logger.log(
+ logger.info(
`No history. startHistoryId: ${startHistoryId}. ${JSON.stringify(decodedData)}`,
);
@@ -199,7 +199,7 @@ export async function processHistoryForUser(
});
}
- logger.log(`Completed. ${JSON.stringify(decodedData)}`);
+ logger.info(`Completed. ${JSON.stringify(decodedData)}`);
return NextResponse.json({ ok: true });
} catch (error) {
@@ -300,13 +300,13 @@ async function processHistoryItem(
});
if (!isFree) {
- logger.log(
+ logger.info(
`Skipping. Message already being processed. email: ${user.email} messageId: ${messageId}`,
);
return;
}
- logger.log(
+ logger.info(
`Getting message... email: ${user.email} messageId: ${messageId} threadId: ${threadId}`,
);
@@ -323,7 +323,7 @@ async function processHistoryItem(
// if the rule has already been executed, skip
if (hasExistingRule) {
- logger.log("Skipping. Rule already exists.");
+ logger.info("Skipping. Rule already exists.");
return;
}
@@ -340,7 +340,7 @@ async function processHistoryItem(
});
if (blocked) {
- logger.log(
+ logger.info(
`Skipping. Blocked unsubscribed email. email: ${user.email} messageId: ${messageId} threadId: ${threadId}`,
);
return;
@@ -355,7 +355,7 @@ async function processHistoryItem(
);
if (shouldRunBlocker) {
- logger.log("Running cold email blocker...");
+ logger.info("Running cold email blocker...");
const hasPreviousEmail = await hasPreviousEmailsFromSenderOrDomain(
gmail,
@@ -395,12 +395,12 @@ async function processHistoryItem(
select: { category: true },
});
if (!existingSender?.category) {
- await categorizeSender(sender, user, gmail, accessToken);
+ await categorizeSender(sender, user, gmail);
}
}
if (hasAutomationRules && hasAiAutomationAccess) {
- logger.log("Running rules...");
+ logger.info("Running rules...");
await runRulesOnMessage({
gmail,
@@ -413,7 +413,7 @@ async function processHistoryItem(
} catch (error: any) {
// gmail bug or snoozed email: https://stackoverflow.com/questions/65290987/gmail-api-getmessage-method-returns-404-for-message-gotten-from-listhistory-meth
if (error.message === "Requested entity was not found.") {
- logger.log(
+ logger.info(
`Message not found. email: ${user.email} messageId: ${messageId} threadId: ${threadId}`,
);
return;
diff --git a/apps/web/app/api/user/categorize/senders/batch/handle-batch-validation.ts b/apps/web/app/api/user/categorize/senders/batch/handle-batch-validation.ts
new file mode 100644
index 0000000000..a6cf7f81b3
--- /dev/null
+++ b/apps/web/app/api/user/categorize/senders/batch/handle-batch-validation.ts
@@ -0,0 +1,7 @@
+import { z } from "zod";
+
+export const aiCategorizeSendersSchema = z.object({
+ userId: z.string(),
+ senders: z.array(z.string()),
+});
+export type AiCategorizeSenders = z.infer
;
diff --git a/apps/web/app/api/user/categorize/senders/batch/handle-batch.ts b/apps/web/app/api/user/categorize/senders/batch/handle-batch.ts
new file mode 100644
index 0000000000..fedadf489f
--- /dev/null
+++ b/apps/web/app/api/user/categorize/senders/batch/handle-batch.ts
@@ -0,0 +1,138 @@
+import { NextResponse } from "next/server";
+import { aiCategorizeSendersSchema } from "@/app/api/user/categorize/senders/batch/handle-batch-validation";
+import { getThreadsFromSender } from "@/utils/gmail/thread";
+import {
+ categorizeWithAi,
+ getCategories,
+ updateSenderCategory,
+} from "@/utils/categorize/senders/categorize";
+import { isDefined } from "@/utils/types";
+import { isActionError } from "@/utils/error";
+import { validateUserAndAiAccess } from "@/utils/user/validate";
+import { getGmailClientWithRefresh } from "@/utils/gmail/client";
+import { UNKNOWN_CATEGORY } from "@/utils/ai/categorize-sender/ai-categorize-senders";
+import { createScopedLogger } from "@/utils/logger";
+import prisma from "@/utils/prisma";
+import { saveCategorizationProgress } from "@/utils/redis/categorization-progress";
+
+const logger = createScopedLogger("api/user/categorize/senders/batch");
+
+export async function handleBatchRequest(
+ request: Request,
+): Promise {
+ try {
+ const handleBatchResult = await handleBatchInternal(request);
+ if (isActionError(handleBatchResult))
+ return NextResponse.json({ error: handleBatchResult.error });
+ return NextResponse.json({ ok: true });
+ } catch (error) {
+ logger.error("handleBatchRequest", { error });
+ return NextResponse.json(
+ { error: "Internal server error" },
+ { status: 500 },
+ );
+ }
+}
+
+async function handleBatchInternal(request: Request) {
+ const json = await request.json();
+ const body = aiCategorizeSendersSchema.parse(json);
+ const { userId, senders } = body;
+
+ logger.trace(`handleBatch ${userId}: ${senders.length} senders`);
+
+ const userResult = await validateUserAndAiAccess(userId);
+ if (isActionError(userResult)) return userResult;
+ const { user } = userResult;
+
+ const categoriesResult = await getCategories(userId);
+ if (isActionError(categoriesResult)) return categoriesResult;
+ const { categories } = categoriesResult;
+
+ const account = await prisma.account.findFirst({
+ where: { user: { id: userId }, provider: "google" },
+ select: {
+ access_token: true,
+ refresh_token: true,
+ expires_at: true,
+ providerAccountId: true,
+ },
+ });
+
+ if (!account) return { error: "No account found" };
+ if (!account.access_token || !account.refresh_token)
+ return { error: "No access or refresh token" };
+
+ const gmail = await getGmailClientWithRefresh(
+ {
+ accessToken: account.access_token,
+ refreshToken: account.refresh_token,
+ expiryDate: account.expires_at,
+ },
+ account.providerAccountId,
+ );
+ if (!gmail) return { error: "No Gmail client" };
+
+ const sendersWithSnippets: Map = new Map();
+
+ // 1. fetch 3 messages for each sender
+ for (const sender of senders) {
+ const threadsFromSender = await getThreadsFromSender(gmail, sender, 3);
+ const snippets = threadsFromSender.map((t) => t.snippet).filter(isDefined);
+ sendersWithSnippets.set(sender, snippets);
+ }
+
+ // 2. categorize senders with ai
+ const results = await categorizeWithAi({
+ user,
+ sendersWithSnippets,
+ categories,
+ });
+
+ // 3. save categorized senders to db
+ for (const result of results) {
+ await updateSenderCategory({
+ sender: result.sender,
+ categories,
+ categoryName: result.category ?? UNKNOWN_CATEGORY,
+ userId,
+ });
+ }
+
+ // // 4. categorize senders that were not categorized
+ // const uncategorizedSenders = results.filter(isUncategorized);
+
+ // await saveCategorizationProgress({
+ // userId,
+ // incrementCompleted: senders.length - uncategorizedSenders.length,
+ // });
+
+ // for (const sender of uncategorizedSenders) {
+ // try {
+ // await categorizeSender(sender.sender, user, gmail, categories);
+ // } catch (error) {
+ // logger.error("Error categorizing sender", {
+ // sender: sender.sender,
+ // error,
+ // });
+ // captureException(error);
+ // }
+
+ // await saveCategorizationProgress({
+ // userId,
+ // incrementCompleted: 1,
+ // });
+ // }
+
+ await saveCategorizationProgress({
+ userId,
+ incrementCompleted: senders.length,
+ });
+
+ return NextResponse.json({ ok: true });
+}
+
+// const isUncategorized = (r: { category?: string }) =>
+// !r.category ||
+// r.category === UNKNOWN_CATEGORY ||
+// r.category === REQUEST_MORE_INFORMATION_CATEGORY;
diff --git a/apps/web/app/api/user/categorize/senders/batch/route.ts b/apps/web/app/api/user/categorize/senders/batch/route.ts
new file mode 100644
index 0000000000..63ad2beafa
--- /dev/null
+++ b/apps/web/app/api/user/categorize/senders/batch/route.ts
@@ -0,0 +1,5 @@
+import { verifySignatureAppRouter } from "@upstash/qstash/nextjs";
+import { withError } from "@/utils/middleware";
+import { handleBatchRequest } from "@/app/api/user/categorize/senders/batch/handle-batch";
+
+export const POST = withError(verifySignatureAppRouter(handleBatchRequest));
diff --git a/apps/web/app/api/user/categorize/senders/batch/simple/route.ts b/apps/web/app/api/user/categorize/senders/batch/simple/route.ts
new file mode 100644
index 0000000000..fc345885db
--- /dev/null
+++ b/apps/web/app/api/user/categorize/senders/batch/simple/route.ts
@@ -0,0 +1,20 @@
+import { NextResponse } from "next/server";
+import { headers } from "next/headers";
+import { withError } from "@/utils/middleware";
+import { handleBatchRequest } from "@/app/api/user/categorize/senders/batch/handle-batch";
+import { env } from "@/env";
+import { isValidInternalApiKey } from "@/utils/internal-api";
+
+// Fallback when Qstash is not in use
+export const POST = withError(async (request) => {
+ if (env.QSTASH_TOKEN) {
+ return NextResponse.json({
+ error: "Qstash is set. This endpoint is disabled.",
+ });
+ }
+
+ if (!isValidInternalApiKey(headers()))
+ return NextResponse.json({ error: "Invalid API key" });
+
+ return handleBatchRequest(request);
+});
diff --git a/apps/web/app/api/user/categorize/senders/find-senders.ts b/apps/web/app/api/user/categorize/senders/find-senders.ts
deleted file mode 100644
index 289824d6a5..0000000000
--- a/apps/web/app/api/user/categorize/senders/find-senders.ts
+++ /dev/null
@@ -1,78 +0,0 @@
-import type { gmail_v1 } from "@googleapis/gmail";
-import { getMessagesBatch } from "@/utils/gmail/message";
-import { getThreadsWithNextPageToken } from "@/utils/gmail/thread";
-import { isDefined } from "@/utils/types";
-import type { SenderMap } from "@/app/api/user/categorize/senders/types";
-
-export async function findSendersWithPagination(
- gmail: gmail_v1.Gmail,
- accessToken: string,
- maxPages: number,
-) {
- const allSenders: SenderMap = new Map();
- let nextPageToken: string | undefined = undefined;
- let currentPage = 0;
-
- while (currentPage < maxPages) {
- const { senders, nextPageToken: newNextPageToken } = await findSenders(
- gmail,
- accessToken,
- nextPageToken,
- );
-
- for (const [sender, messages] of Object.entries(senders)) {
- const existingMessages = allSenders.get(sender) ?? [];
- allSenders.set(sender, [...existingMessages, ...messages]);
- }
-
- if (!newNextPageToken) break; // No more pages
-
- nextPageToken = newNextPageToken;
- currentPage++;
- }
-
- return { senders: allSenders, nextPageToken };
-}
-
-export async function findSenders(
- gmail: gmail_v1.Gmail,
- accessToken: string,
- pageToken?: string,
- maxResults = 50,
-) {
- const senders: SenderMap = new Map();
-
- const { threads, nextPageToken } = await getThreadsWithNextPageToken({
- q: "-in:sent",
- gmail,
- maxResults,
- pageToken,
- });
-
- const messageIds = threads.map((t) => t.id).filter(isDefined);
- const messages = await getMessagesBatch(messageIds, accessToken);
-
- for (const message of messages) {
- const sender = message.headers.from;
- if (sender) {
- const existingMessages = senders.get(sender) ?? [];
- senders.set(sender, [...existingMessages, message]);
- }
- }
-
- return { senders, nextPageToken };
-}
-
-function isNotFoundError(error: unknown): boolean {
- return (
- typeof error === "object" &&
- error !== null &&
- "errors" in error &&
- Array.isArray((error as any).errors) &&
- (error as any).errors.some(
- (e: any) =>
- e.message === "Requested entity was not found." &&
- e.reason === "notFound",
- )
- );
-}
diff --git a/apps/web/app/api/user/categorize/senders/progress/route.ts b/apps/web/app/api/user/categorize/senders/progress/route.ts
new file mode 100644
index 0000000000..9ad94fb887
--- /dev/null
+++ b/apps/web/app/api/user/categorize/senders/progress/route.ts
@@ -0,0 +1,21 @@
+import { NextResponse } from "next/server";
+import { auth } from "@/app/api/auth/[...nextauth]/auth";
+import { getCategorizationProgress } from "@/utils/redis/categorization-progress";
+import { withError } from "@/utils/middleware";
+
+export type CategorizeProgress = Awaited<
+ ReturnType
+>;
+
+async function getCategorizeProgress(userId: string) {
+ const progress = await getCategorizationProgress({ userId });
+ return progress;
+}
+
+export const GET = withError(async () => {
+ const session = await auth();
+ if (!session?.user) return NextResponse.json({ error: "Not authenticated" });
+
+ const result = await getCategorizeProgress(session.user.id);
+ return NextResponse.json(result);
+});
diff --git a/apps/web/app/api/user/categorize/senders/uncategorized/get-uncategorized-senders.ts b/apps/web/app/api/user/categorize/senders/uncategorized/get-uncategorized-senders.ts
new file mode 100644
index 0000000000..c029e4f2a1
--- /dev/null
+++ b/apps/web/app/api/user/categorize/senders/uncategorized/get-uncategorized-senders.ts
@@ -0,0 +1,55 @@
+import { getSenders } from "@inboxzero/tinybird";
+import prisma from "@/utils/prisma";
+
+const MAX_ITERATIONS = 200;
+
+export async function getUncategorizedSenders({
+ email,
+ userId,
+ offset = 0,
+ limit = 100,
+}: {
+ email: string;
+ userId: string;
+ offset?: number;
+ limit?: number;
+}) {
+ let uncategorizedSenders: string[] = [];
+ let currentOffset = offset;
+
+ while (uncategorizedSenders.length === 0 && currentOffset < MAX_ITERATIONS) {
+ const result = await getSenders({
+ ownerEmail: email,
+ limit,
+ offset: currentOffset,
+ });
+ const allSenders = result.data.map((sender) => sender.from);
+
+ const existingSenders = await prisma.newsletter.findMany({
+ where: {
+ email: { in: allSenders },
+ userId,
+ category: { isNot: null },
+ },
+ select: { email: true },
+ });
+
+ const existingSenderEmails = new Set(existingSenders.map((s) => s.email));
+
+ uncategorizedSenders = allSenders.filter(
+ (email) => !existingSenderEmails.has(email),
+ );
+
+ // Break the loop if no more senders are available
+ if (allSenders.length < limit) {
+ return { uncategorizedSenders };
+ }
+
+ currentOffset += limit;
+ }
+
+ return {
+ uncategorizedSenders,
+ nextOffset: currentOffset, // Only return nextOffset if there might be more
+ };
+}
diff --git a/apps/web/app/api/user/categorize/senders/uncategorized/route.ts b/apps/web/app/api/user/categorize/senders/uncategorized/route.ts
index 4881d474fe..e56d849c6a 100644
--- a/apps/web/app/api/user/categorize/senders/uncategorized/route.ts
+++ b/apps/web/app/api/user/categorize/senders/uncategorized/route.ts
@@ -1,72 +1,19 @@
import { NextResponse } from "next/server";
import { withError } from "@/utils/middleware";
import { getSessionAndGmailClient } from "@/utils/actions/helpers";
-import { getSenders } from "@inboxzero/tinybird";
-import prisma from "@/utils/prisma";
+import { isActionError } from "@/utils/error";
+import { getUncategorizedSenders } from "@/app/api/user/categorize/senders/uncategorized/get-uncategorized-senders";
export type UncategorizedSendersResponse = {
uncategorizedSenders: string[];
nextOffset?: number;
};
-async function getUncategorizedSenders({
- email,
- userId,
- offset = 0,
- limit = 100,
-}: {
- email: string;
- userId: string;
- offset?: number;
- limit?: number;
-}) {
- let uncategorizedSenders: string[] = [];
- let currentOffset = offset;
-
- while (uncategorizedSenders.length === 0) {
- const result = await getSenders({
- ownerEmail: email,
- limit,
- offset: currentOffset,
- });
- const allSenders = result.data.map((sender) => sender.from);
-
- const existingSenders = await prisma.newsletter.findMany({
- where: {
- email: { in: allSenders },
- userId,
- category: { isNot: null },
- },
- select: { email: true },
- });
-
- const existingSenderEmails = new Set(existingSenders.map((s) => s.email));
-
- uncategorizedSenders = allSenders.filter(
- (email) => !existingSenderEmails.has(email),
- );
-
- // Break the loop if no more senders are available
- if (allSenders.length < limit) {
- return { uncategorizedSenders };
- }
-
- currentOffset += limit;
- }
-
- return {
- uncategorizedSenders,
- nextOffset: currentOffset, // Only return nextOffset if there might be more
- };
-}
-
export const GET = withError(async (request: Request) => {
- const { gmail, user, error, session } = await getSessionAndGmailClient();
- if (!user?.email) return NextResponse.json({ error: "Not authenticated" });
- if (error) return NextResponse.json({ error });
- if (!gmail) return NextResponse.json({ error: "Could not load Gmail" });
- if (!session?.accessToken)
- return NextResponse.json({ error: "No access token" });
+ const sessionResult = await getSessionAndGmailClient();
+ if (isActionError(sessionResult))
+ return NextResponse.json({ error: sessionResult.error });
+ const { user } = sessionResult;
const url = new URL(request.url);
const offset = Number.parseInt(url.searchParams.get("offset") || "0");
diff --git a/apps/web/components/ProgressPanel.tsx b/apps/web/components/ProgressPanel.tsx
new file mode 100644
index 0000000000..dc0e733daf
--- /dev/null
+++ b/apps/web/components/ProgressPanel.tsx
@@ -0,0 +1,57 @@
+"use client";
+
+import { AnimatePresence, motion } from "framer-motion";
+import { ProgressBar } from "@tremor/react";
+import { cn } from "@/utils";
+
+export function ProgressPanel({
+ totalItems,
+ remainingItems,
+ inProgressText,
+ completedText,
+ itemLabel,
+}: {
+ totalItems: number;
+ remainingItems: number;
+ inProgressText: string;
+ completedText: string;
+ itemLabel: string;
+}) {
+ const totalProcessed = totalItems - remainingItems;
+ const progress = (totalProcessed / totalItems) * 100;
+ const isCompleted = progress === 100;
+
+ if (!totalItems) return null;
+
+ return (
+
+
+
+
+
+
+ {isCompleted ? completedText : inProgressText}
+
+
+ {totalProcessed} of {totalItems} {itemLabel} processed
+
+
+
+
+