diff --git a/apps/web/__tests__/ai-categorize-senders.test.ts b/apps/web/__tests__/ai-categorize-senders.test.ts index 8e66822d93..ddac02a2c9 100644 --- a/apps/web/__tests__/ai-categorize-senders.test.ts +++ b/apps/web/__tests__/ai-categorize-senders.test.ts @@ -1,8 +1,5 @@ import { describe, it, expect, vi } from "vitest"; -import { - aiCategorizeSenders, - REQUEST_MORE_INFORMATION_CATEGORY, -} from "@/utils/ai/categorize-sender/ai-categorize-senders"; +import { aiCategorizeSenders } from "@/utils/ai/categorize-sender/ai-categorize-senders"; import { defaultCategory } from "@/utils/categories"; import { aiCategorizeSender } from "@/utils/ai/categorize-sender/ai-categorize-single-sender"; import { getEmailAccount } from "@/__tests__/helpers"; @@ -135,9 +132,7 @@ describe.runIf(isAiTest)("AI Sender Categorization", () => { }); if (expectedCategory === "Unknown") { - expect([REQUEST_MORE_INFORMATION_CATEGORY, "Unknown"]).toContain( - result?.category, - ); + expect(result?.category).toBe("Unknown"); } else { expect(result?.category).toBe(expectedCategory); } @@ -161,9 +156,7 @@ describe.runIf(isAiTest)("AI Sender Categorization", () => { categories: getEnabledCategories(), }); - expect([REQUEST_MORE_INFORMATION_CATEGORY, "Unknown"]).toContain( - result?.category, - ); + expect(result?.category).toBe("Unknown"); }, TIMEOUT, ); diff --git a/apps/web/app/(app)/[emailAccountId]/deep-clean/BulkOperationProgress.tsx b/apps/web/app/(app)/[emailAccountId]/deep-clean/BulkOperationProgress.tsx new file mode 100644 index 0000000000..91bce6e9c8 --- /dev/null +++ b/apps/web/app/(app)/[emailAccountId]/deep-clean/BulkOperationProgress.tsx @@ -0,0 +1,63 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { ProgressPanel } from "@/components/ProgressPanel"; +import { useBulkOperationProgress } from "@/hooks/useDeepClean"; + +export function BulkOperationProgress() { + const [hasActiveOperations, setHasActiveOperations] = useState(false); + + const { data } = useBulkOperationProgress( + hasActiveOperations ? 2000 : 10_000, // Poll more frequently when operations are active + ); + + const operations = data?.operations || []; + const activeOperations = operations.filter( + (op) => op.status === "processing" || op.status === "pending", + ); + + useEffect(() => { + setHasActiveOperations(activeOperations.length > 0); + }, [activeOperations.length]); + + // Show progress for each active operation + return ( + <> + {operations.map((operation) => { + const isActive = + operation.status === "processing" || operation.status === "pending"; + const isCompleted = operation.status === "completed"; + + if (!isActive && !isCompleted) return null; + + // Hide completed operations after 5 seconds + if (isCompleted) { + setTimeout(() => { + // This will be handled by React's re-render when the operation is removed from Redis + }, 5000); + } + + const displayName = + operation.operationType === "archive" + ? `Archiving ${operation.categoryOrSender}` + : `Marking ${operation.categoryOrSender} as read`; + + const completedText = + operation.operationType === "archive" + ? `Archived ${operation.completedItems} emails!` + : `Marked ${operation.completedItems} emails as read!`; + + return ( + + ); + })} + + ); +} diff --git a/apps/web/app/(app)/[emailAccountId]/deep-clean/CategorizeProgress.tsx b/apps/web/app/(app)/[emailAccountId]/deep-clean/CategorizeProgress.tsx new file mode 100644 index 0000000000..111bda636a --- /dev/null +++ b/apps/web/app/(app)/[emailAccountId]/deep-clean/CategorizeProgress.tsx @@ -0,0 +1,80 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { atom, useAtom } from "jotai"; +import useSWR from "swr"; +import { ProgressPanel } from "@/components/ProgressPanel"; +import type { CategorizeProgress } from "@/app/api/user/categorize/senders/progress/route"; +import { useInterval } from "@/hooks/useInterval"; + +const isCategorizeInProgressAtom = atom(false); + +export function useCategorizeProgress() { + const [isBulkCategorizing, setIsBulkCategorizing] = useAtom( + isCategorizeInProgressAtom, + ); + return { isBulkCategorizing, setIsBulkCategorizing }; +} + +export function CategorizeSendersProgress({ + refresh = false, +}: { + refresh: boolean; +}) { + const { isBulkCategorizing } = useCategorizeProgress(); + const [fakeProgress, setFakeProgress] = useState(0); + + const { data } = useSWR( + "/api/user/categorize/senders/progress", + { + refreshInterval: refresh || isBulkCategorizing ? 1000 : undefined, + }, + ); + + useInterval( + () => { + if (!data?.totalItems) return; + + setFakeProgress((prev) => { + const realCompleted = data.completedItems || 0; + if (realCompleted > prev) return realCompleted; + + const maxProgress = Math.min( + Math.floor(data.totalItems * 0.9), + realCompleted + 30, + ); + return prev < maxProgress ? prev + 1 : prev; + }); + }, + isBulkCategorizing ? 1500 : null, + ); + + const { setIsBulkCategorizing } = useCategorizeProgress(); + useEffect(() => { + let timeoutId: NodeJS.Timeout | undefined; + if (data?.completedItems === data?.totalItems) { + timeoutId = setTimeout(() => { + setIsBulkCategorizing(false); + setFakeProgress(0); + }, 3000); + } + return () => { + if (timeoutId) clearTimeout(timeoutId); + }; + }, [data?.completedItems, data?.totalItems, setIsBulkCategorizing]); + + if (!data) return null; + + const totalItems = data.totalItems || 0; + const displayedProgress = Math.max(data.completedItems || 0, fakeProgress); + + return ( + + ); +} diff --git a/apps/web/app/(app)/[emailAccountId]/deep-clean/CategorizeWithAiButton.tsx b/apps/web/app/(app)/[emailAccountId]/deep-clean/CategorizeWithAiButton.tsx new file mode 100644 index 0000000000..8ed11fce1d --- /dev/null +++ b/apps/web/app/(app)/[emailAccountId]/deep-clean/CategorizeWithAiButton.tsx @@ -0,0 +1,105 @@ +"use client"; + +import { useState } from "react"; +import { SparklesIcon } from "lucide-react"; +import { toast } from "sonner"; +import { Button } from "@/components/ui/button"; +import { bulkCategorizeSendersAction } from "@/utils/actions/categorize"; +import { PremiumTooltip, usePremium } from "@/components/PremiumAlert"; +import { usePremiumModal } from "@/app/(app)/premium/PremiumModal"; +import type { ButtonProps } from "@/components/ui/button"; +import { useCategorizeProgress } from "@/app/(app)/[emailAccountId]/smart-categories/CategorizeProgress"; +import { Tooltip } from "@/components/Tooltip"; +import { useAccount } from "@/providers/EmailAccountProvider"; + +export function CategorizeWithAiButton({ + buttonProps, +}: { + buttonProps?: ButtonProps; +}) { + const { emailAccountId } = useAccount(); + const [isCategorizing, setIsCategorizing] = useState(false); + const { hasAiAccess } = usePremium(); + const { PremiumModal, openModal: openPremiumModal } = usePremiumModal(); + + const { setIsBulkCategorizing } = useCategorizeProgress(); + + return ( + <> + + + + + + ); +} + +function CategorizeWithAiButtonTooltip({ + children, + hasAiAccess, + openPremiumModal, +}: { + children: React.ReactElement; + hasAiAccess: boolean; + openPremiumModal: () => void; +}) { + if (hasAiAccess) { + return ( + + {children} + + ); + } + + return ( + + {children} + + ); +} diff --git a/apps/web/app/(app)/[emailAccountId]/deep-clean/CreateCategoryButton.tsx b/apps/web/app/(app)/[emailAccountId]/deep-clean/CreateCategoryButton.tsx new file mode 100644 index 0000000000..7150d6bfc3 --- /dev/null +++ b/apps/web/app/(app)/[emailAccountId]/deep-clean/CreateCategoryButton.tsx @@ -0,0 +1,237 @@ +"use client"; + +import { useCallback } from "react"; +import { type SubmitHandler, useForm } from "react-hook-form"; +import { zodResolver } from "@hookform/resolvers/zod"; +import { PlusIcon } from "lucide-react"; +import { useModal } from "@/hooks/useModal"; +import { Button, type ButtonProps } from "@/components/ui/button"; +import { Input } from "@/components/Input"; +import { toastSuccess, toastError } from "@/components/Toast"; +import { + createCategoryBody, + type CreateCategoryBody, +} from "@/utils/actions/categorize.validation"; +import { createCategoryAction } from "@/utils/actions/categorize"; +import { + Dialog, + DialogContent, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import type { Category } from "@prisma/client"; +import { MessageText } from "@/components/Typography"; +import { useAccount } from "@/providers/EmailAccountProvider"; + +type ExampleCategory = { + name: string; + description: string; +}; + +const EXAMPLE_CATEGORIES: ExampleCategory[] = [ + { + name: "Team", + description: + "Internal team members with @company.com email addresses, including employees and colleagues within our organization", + }, + { + name: "Customer", + description: + "Email addresses belonging to customers, including those reaching out for support or engaging with customer success", + }, + { + name: "Candidate", + description: + "Job applicants, potential hires, and candidates in your interview pipeline", + }, + { + name: "Job Application", + description: + "Companies, hiring platforms, and recruiters you've applied to or are interviewing with for positions", + }, + { + name: "Investor", + description: + "Current and potential investors, investment firms, and venture capital contacts", + }, + { + name: "Founder", + description: + "Startup founders, entrepreneurs, and potential portfolio companies seeking investment or partnerships", + }, + { + name: "Vendor", + description: + "Service providers, suppliers, and business partners who provide products or services to your company", + }, + { + name: "Server Error", + description: "Automated monitoring services and error reporting systems", + }, + { + name: "Press", + description: + "Journalists, media outlets, PR agencies, and industry publications seeking interviews or coverage", + }, + { + name: "Conference", + description: + "Event organizers, conference coordinators, and speaking opportunity contacts for industry events", + }, + { + name: "Nonprofit", + description: + "Charitable organizations, NGOs, social impact organizations, and philanthropic foundations", + }, +]; + +export function CreateCategoryButton({ + buttonProps, +}: { + buttonProps?: ButtonProps; +}) { + const { isModalOpen, openModal, closeModal, setIsModalOpen } = useModal(); + + return ( +
+ + + +
+ ); +} + +export function CreateCategoryDialog({ + category, + isOpen, + onOpenChange, + closeModal, +}: { + category?: Pick; + isOpen: boolean; + onOpenChange: (open: boolean) => void; + closeModal: () => void; +}) { + return ( + + + + Create Category + + + + + + ); +} + +function CreateCategoryForm({ + category, + closeModal, +}: { + category?: Pick & { id?: string }; + closeModal: () => void; +}) { + const { emailAccountId } = useAccount(); + + const { + register, + handleSubmit, + formState: { errors, isSubmitting }, + setValue, + } = useForm({ + resolver: zodResolver(createCategoryBody), + defaultValues: { + id: category?.id, + name: category?.name, + description: category?.description, + }, + }); + + const handleExampleClick = useCallback( + (category: ExampleCategory) => { + setValue("name", category.name); + setValue("description", category.description); + }, + [setValue], + ); + + const onSubmit: SubmitHandler = useCallback( + async (data) => { + const result = await createCategoryAction(emailAccountId, data); + + if (result?.serverError) { + toastError({ + description: `There was an error creating the category. ${result.serverError || ""}`, + }); + } else { + toastSuccess({ description: "Category created!" }); + closeModal(); + } + }, + [closeModal, emailAccountId], + ); + + return ( +
+ + + +
+
Examples
+
+ {EXAMPLE_CATEGORIES.map((category) => ( + + ))} +
+
+ + {category && ( + + Note: editing a category name/description only impacts future + categorization. Existing email addresses in this category will not be + affected. + + )} + + +
+ ); +} diff --git a/apps/web/app/(app)/[emailAccountId]/deep-clean/DeepCleanContent.tsx b/apps/web/app/(app)/[emailAccountId]/deep-clean/DeepCleanContent.tsx new file mode 100644 index 0000000000..19f7162f82 --- /dev/null +++ b/apps/web/app/(app)/[emailAccountId]/deep-clean/DeepCleanContent.tsx @@ -0,0 +1,106 @@ +"use client"; + +import { useCallback, useRef, useEffect } from "react"; +import { ClientOnly } from "@/components/ClientOnly"; +import { CategorizeSendersProgress } from "@/app/(app)/[emailAccountId]/smart-categories/CategorizeProgress"; +import { BulkOperationProgress } from "@/app/(app)/[emailAccountId]/deep-clean/BulkOperationProgress"; +import { useAccount } from "@/providers/EmailAccountProvider"; +import { Button } from "@/components/ui/button"; +import { RefreshCw } from "lucide-react"; +import { categorizeMoreSendersAction } from "@/utils/actions/deep-clean"; +import { toast } from "sonner"; +import { LoadingContent } from "@/components/LoadingContent"; +import { DeepCleanGroupedTable } from "@/components/DeepCleanGroupedTable"; +import { useDeepCleanSenders } from "@/hooks/useDeepClean"; +import { PageHeader } from "@/components/PageHeader"; +import { PageWrapper } from "@/components/PageWrapper"; + +export function DeepCleanContent() { + const { emailAccount } = useAccount(); + const { data, isLoading, error, mutate } = useDeepCleanSenders(); + const hasAutoTriggered = useRef(false); + + const handleCategorizeMore = useCallback(async () => { + if (!emailAccount?.id) return; + + try { + const result = await categorizeMoreSendersAction(emailAccount.id, { + limit: 100, + }); + + if (result?.data?.success) { + toast.success(result.data.message); + // Refresh data after categorization completes + setTimeout(() => { + mutate(); + }, 2000); + } else { + toast.error(result?.serverError || "Failed to categorize senders"); + } + } catch (error) { + toast.error("Failed to categorize senders"); + console.error("Categorize more error:", error); + } + }, [emailAccount?.id, mutate]); + + // Auto-trigger categorization on first load if we have very few senders + useEffect(() => { + if (hasAutoTriggered.current || !data || isLoading) return; + + // If first-time user with few/no senders, automatically fetch and categorize + if (data.senders.length < 5) { + hasAutoTriggered.current = true; + handleCategorizeMore(); + } + }, [data, isLoading, handleCategorizeMore]); + + if (!data || data.senders.length === 0) { + return ( + +
+
+

+ No categorized senders found. Start categorizing your inbox to use + DeepClean. +

+ +
+
+ + ); + } + + return ( + + +
+ + + +
+ + + + + + + + ({ + address: sender.email, + category: sender.category, + meta: { width: "auto" }, + }))} + categories={data.categories} + /> + + ); +} diff --git a/apps/web/app/(app)/[emailAccountId]/deep-clean/Uncategorized.tsx b/apps/web/app/(app)/[emailAccountId]/deep-clean/Uncategorized.tsx new file mode 100644 index 0000000000..02f5c62b54 --- /dev/null +++ b/apps/web/app/(app)/[emailAccountId]/deep-clean/Uncategorized.tsx @@ -0,0 +1,198 @@ +"use client"; + +import useSWRInfinite from "swr/infinite"; +import { useMemo, useCallback } from "react"; +import { ChevronsDownIcon, SparklesIcon, StopCircleIcon } from "lucide-react"; +import { ClientOnly } from "@/components/ClientOnly"; +import { SendersTable } from "@/components/GroupedTable"; +import { LoadingContent } from "@/components/LoadingContent"; +import { Button } from "@/components/ui/button"; +import type { UncategorizedSendersResponse } from "@/app/api/user/categorize/senders/uncategorized/route"; +import { TopBar } from "@/components/TopBar"; +import { toastError } from "@/components/Toast"; +import { + useHasProcessingItems, + pushToAiCategorizeSenderQueueAtom, + stopAiCategorizeSenderQueue, +} from "@/store/ai-categorize-sender-queue"; +import { SectionDescription } from "@/components/Typography"; +import { ButtonLoader } from "@/components/Loading"; +import { PremiumTooltip, usePremium } from "@/components/PremiumAlert"; +import { usePremiumModal } from "@/app/(app)/premium/PremiumModal"; +import { Toggle } from "@/components/Toggle"; +import { setAutoCategorizeAction } from "@/utils/actions/categorize"; +import { TooltipExplanation } from "@/components/TooltipExplanation"; +import type { CategoryWithRules } from "@/utils/category.server"; +import { useAccount } from "@/providers/EmailAccountProvider"; + +export function Uncategorized({ + categories, + autoCategorizeSenders, +}: { + categories: CategoryWithRules[]; + autoCategorizeSenders: boolean; +}) { + const { hasAiAccess } = usePremium(); + const { PremiumModal, openModal: openPremiumModal } = usePremiumModal(); + + const { data: senderAddresses, loadMore, isLoading, hasMore } = useSenders(); + const hasProcessingItems = useHasProcessingItems(); + + const senders = useMemo( + () => + senderAddresses?.map((address) => { + return { address, category: null }; + }), + [senderAddresses], + ); + + const { emailAccountId } = useAccount(); + + return ( + + +
+ + + + + {hasProcessingItems && ( + + )} +
+ +
+
+ +
+ +
+ + + {senders?.length ? ( + <> + + {hasMore && ( + + )} + + ) : ( + !isLoading && ( + + No senders left to categorize! + + ) + )} + + + + ); +} + +function AutoCategorizeToggle({ + autoCategorizeSenders, + emailAccountId, +}: { + autoCategorizeSenders: boolean; + emailAccountId: string; +}) { + return ( + { + await setAutoCategorizeAction(emailAccountId, { + autoCategorizeSenders: enabled, + }); + }} + /> + ); +} + +function useSenders() { + const getKey = ( + pageIndex: number, + previousPageData: UncategorizedSendersResponse | null, + ) => { + // Reached the end + if (previousPageData && !previousPageData.nextOffset) return null; + + const baseUrl = "/api/user/categorize/senders/uncategorized"; + const offset = pageIndex === 0 ? 0 : previousPageData?.nextOffset; + + return `${baseUrl}?offset=${offset}`; + }; + + const { data, size, setSize, isLoading } = + useSWRInfinite(getKey, { + revalidateOnFocus: false, + revalidateFirstPage: false, + persistSize: true, + revalidateOnMount: true, + }); + + const loadMore = useCallback(() => { + setSize(size + 1); + }, [setSize, size]); + + // Combine all senders from all pages + const allSenders = useMemo(() => { + return data?.flatMap((page) => page.uncategorizedSenders); + }, [data]); + + // Check if there's more data to load by looking at the last page + const hasMore = !!data?.[data.length - 1]?.nextOffset; + + return { + data: allSenders, + loadMore, + isLoading, + hasMore, + }; +} diff --git a/apps/web/app/(app)/[emailAccountId]/deep-clean/page.tsx b/apps/web/app/(app)/[emailAccountId]/deep-clean/page.tsx new file mode 100644 index 0000000000..efa76df2ec --- /dev/null +++ b/apps/web/app/(app)/[emailAccountId]/deep-clean/page.tsx @@ -0,0 +1,11 @@ +import { DeepCleanContent } from "@/app/(app)/[emailAccountId]/deep-clean/DeepCleanContent"; +import { PermissionsCheck } from "@/app/(app)/[emailAccountId]/PermissionsCheck"; + +export default async function CategoriesPage() { + return ( + <> + + + + ); +} 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 index 96ce5169ab..2aefa1b107 100644 --- a/apps/web/app/api/user/categorize/senders/batch/handle-batch.ts +++ b/apps/web/app/api/user/categorize/senders/batch/handle-batch.ts @@ -8,7 +8,6 @@ import { } from "@/utils/categorize/senders/categorize"; 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"; @@ -44,7 +43,7 @@ async function handleBatchInternal(request: Request) { const userResult = await validateUserAndAiAccess({ emailAccountId }); const { emailAccount } = userResult; - const categoriesResult = await getCategories({ emailAccountId }); + const categoriesResult = await getCategories(); const { categories } = categoriesResult; const emailAccountWithAccount = await prisma.emailAccount.findUnique({ @@ -103,8 +102,9 @@ async function handleBatchInternal(request: Request) { await updateSenderCategory({ sender: result.sender, categories, - categoryName: result.category ?? UNKNOWN_CATEGORY, + categoryName: result.category ?? "Unknown", emailAccountId, + priority: "priority" in result ? result.priority : undefined, }); } diff --git a/apps/web/app/api/user/deep-clean/progress/route.ts b/apps/web/app/api/user/deep-clean/progress/route.ts new file mode 100644 index 0000000000..270d682dbf --- /dev/null +++ b/apps/web/app/api/user/deep-clean/progress/route.ts @@ -0,0 +1,55 @@ +import { NextResponse } from "next/server"; +import { withEmailAccount } from "@/utils/middleware"; +import { createScopedLogger } from "@/utils/logger"; +import { + getBulkOperationProgress, + getAllBulkOperations, +} from "@/utils/redis/bulk-operation-progress"; + +const logger = createScopedLogger("api/user/deep-clean/progress"); + +export type BulkOperationProgress = Awaited< + ReturnType +>; + +async function getBulkOperationProgressInternal({ + emailAccountId, + operationId, +}: { + emailAccountId: string; + operationId?: string; +}) { + // If operationId is provided, get specific operation + if (operationId) { + const progress = await getBulkOperationProgress({ + emailAccountId, + operationId, + }); + + return progress ? [{ ...progress, operationId }] : []; + } + + // Otherwise, get all operations for this user + return await getAllBulkOperations({ emailAccountId }); +} + +export const GET = withEmailAccount(async (request) => { + const { emailAccountId } = request.auth; + const { searchParams } = new URL(request.url); + const operationId = searchParams.get("operationId") || undefined; + + const operations = await getBulkOperationProgressInternal({ + emailAccountId, + operationId, + }); + + logger.info("Fetched bulk operation progress", { + emailAccountId, + operationId, + operationCount: operations.length, + }); + + return NextResponse.json({ + operations, + }); +}); diff --git a/apps/web/app/api/user/deep-clean/senders/route.ts b/apps/web/app/api/user/deep-clean/senders/route.ts new file mode 100644 index 0000000000..ebd0d19588 --- /dev/null +++ b/apps/web/app/api/user/deep-clean/senders/route.ts @@ -0,0 +1,93 @@ +import { NextResponse } from "next/server"; +import { withEmailAccount } from "@/utils/middleware"; +import prisma from "@/utils/prisma"; +import { createScopedLogger } from "@/utils/logger"; +import { numberToPriority } from "@/utils/priority"; +import { getUserCategoriesWithRules } from "@/utils/category.server"; + +const logger = createScopedLogger("api/user/deep-clean/senders"); + +export type DeepCleanSendersResponse = Awaited< + ReturnType +>; + +async function getDeepCleanSenders({ + emailAccountId, +}: { + emailAccountId: string; +}) { + const categories = await getUserCategoriesWithRules({ emailAccountId }); + + const categorizedSenders = await prisma.newsletter.findMany({ + where: { + emailAccountId, + categoryId: { not: null }, + }, + include: { + category: { + select: { + name: true, + }, + }, + }, + orderBy: { + priority: "desc", // Higher priority first + }, + }); + + // Get email counts for each sender + const senderEmails = await prisma.emailMessage.groupBy({ + by: ["from"], + where: { + emailAccountId, + inbox: true, + from: { + in: categorizedSenders.map((s) => s.email), + }, + }, + _count: { + from: true, + }, + }); + + const emailCountMap = new Map( + senderEmails.map((item) => [item.from, item._count.from]), + ); + + // Create senders array with category information + const senders = categorizedSenders.map((sender) => { + const emailCount = emailCountMap.get(sender.email) || 0; + const category = categories.find((c) => c.id === sender.categoryId) || null; + + return { + email: sender.email, + priority: sender.priority ? numberToPriority(sender.priority) : null, + emailCount, + category, + }; + }); + + // Sort senders by email count (descending) + senders.sort((a, b) => b.emailCount - a.emailCount); + + logger.info("Fetched DeepClean senders", { + emailAccountId, + totalCategories: categories.length, + totalSenders: senders.length, + }); + + return { + senders, + categories, + totalCategorizedSenders: senders.length, + }; +} + +export const GET = withEmailAccount(async (request) => { + const { emailAccountId } = request.auth; + + logger.info("Fetching DeepClean senders", { emailAccountId }); + + const data = await getDeepCleanSenders({ emailAccountId }); + return NextResponse.json(data); +}); diff --git a/apps/web/app/api/webhooks/deep-clean/archive/route.ts b/apps/web/app/api/webhooks/deep-clean/archive/route.ts new file mode 100644 index 0000000000..cf7947c719 --- /dev/null +++ b/apps/web/app/api/webhooks/deep-clean/archive/route.ts @@ -0,0 +1,187 @@ +import { NextResponse } from "next/server"; +import { validateUserAndAiAccess } from "@/utils/user/validate"; +import { createScopedLogger } from "@/utils/logger"; +import { SafeError } from "@/utils/error"; +import { createEmailProvider } from "@/utils/email/provider"; +import prisma from "@/utils/prisma"; +import { + createBulkOperation, + updateBulkOperationProgress, +} from "@/utils/redis/bulk-operation-progress"; + +const logger = createScopedLogger("api/webhooks/deep-clean/archive"); + +/** + * Webhook handler called by QStash to process archive operations. + * This is NOT a user-facing mutation - users call archiveCategoryAction which queues this job. + */ +export async function POST(request: Request) { + try { + const body = await request.json(); + const { emailAccountId, operationId, category, senders } = body; + + if (!emailAccountId || !operationId || !category || !senders) { + return NextResponse.json( + { error: "Missing required fields" }, + { status: 400 }, + ); + } + + logger.info("Starting archive operation", { + emailAccountId, + operationId, + category, + senderCount: senders.length, + }); + + // Validate user access + await validateUserAndAiAccess({ emailAccountId }); + + // Get email account details + const emailAccountWithAccount = await prisma.emailAccount.findUnique({ + where: { id: emailAccountId }, + select: { + email: true, + account: { + select: { + provider: true, + }, + }, + }, + }); + + const account = emailAccountWithAccount?.account; + const ownerEmail = emailAccountWithAccount?.email; + if (!account) throw new SafeError("No account found"); + if (!ownerEmail) throw new SafeError("No email found"); + + // Create email provider (works for both Gmail and Outlook) + const emailProvider = await createEmailProvider({ + emailAccountId, + provider: account.provider, + }); + + // Get all thread IDs for these senders + const allThreadIds: string[] = []; + for (const sender of senders) { + try { + const threads = await emailProvider.getThreadsFromSenderWithSubject( + sender, + 1000, + ); + allThreadIds.push(...threads.map((t) => t.id)); + } catch (error) { + logger.warn("Failed to get threads for sender", { + sender, + error, + }); + // Continue with other senders even if one fails + } + } + + if (allThreadIds.length === 0) { + logger.info("No threads found to archive", { + emailAccountId, + operationId, + }); + return NextResponse.json({ + ok: true, + message: "No threads found to archive", + operationId, + }); + } + + // Create progress tracker + await createBulkOperation({ + emailAccountId, + operationId, + operationType: "archive", + categoryOrSender: category, + totalItems: allThreadIds.length, + }); + + // Update status to processing + await updateBulkOperationProgress({ + emailAccountId, + operationId, + status: "processing", + }); + + let successCount = 0; + let errorCount = 0; + + // Process threads in batches + const BATCH_SIZE = 50; + for (let i = 0; i < allThreadIds.length; i += BATCH_SIZE) { + const batch = allThreadIds.slice(i, i + BATCH_SIZE); + + await Promise.all( + batch.map(async (threadId) => { + try { + await emailProvider.archiveThread(threadId, ownerEmail); + successCount++; + + // Update progress every 10 threads + if (successCount % 10 === 0) { + await updateBulkOperationProgress({ + emailAccountId, + operationId, + incrementCompleted: 10, + }); + } + } catch (error) { + errorCount++; + logger.warn("Failed to archive thread", { + threadId, + error, + }); + } + }), + ); + + logger.info("Batch archived", { + operationId, + batchNumber: Math.floor(i / BATCH_SIZE) + 1, + totalBatches: Math.ceil(allThreadIds.length / BATCH_SIZE), + successCount, + errorCount, + }); + } + + // Final progress update + await updateBulkOperationProgress({ + emailAccountId, + operationId, + incrementCompleted: successCount % 10, // Update any remaining + incrementFailed: errorCount, + status: "completed", + }); + + logger.info("Archive operation completed", { + emailAccountId, + operationId, + totalThreads: allThreadIds.length, + successCount, + errorCount, + }); + + return NextResponse.json({ + ok: true, + message: `Archived ${successCount} emails${errorCount > 0 ? ` (${errorCount} failed)` : ""}`, + operationId, + successCount, + errorCount, + }); + } catch (error) { + logger.error("Archive operation error", { error }); + + if (error instanceof SafeError) { + return NextResponse.json({ error: error.message }, { status: 400 }); + } + + return NextResponse.json( + { error: "Internal server error" }, + { status: 500 }, + ); + } +} diff --git a/apps/web/app/api/webhooks/deep-clean/mark-read/route.ts b/apps/web/app/api/webhooks/deep-clean/mark-read/route.ts new file mode 100644 index 0000000000..0b7710dc42 --- /dev/null +++ b/apps/web/app/api/webhooks/deep-clean/mark-read/route.ts @@ -0,0 +1,187 @@ +import { NextResponse } from "next/server"; +import { validateUserAndAiAccess } from "@/utils/user/validate"; +import { createScopedLogger } from "@/utils/logger"; +import { SafeError } from "@/utils/error"; +import { createEmailProvider } from "@/utils/email/provider"; +import prisma from "@/utils/prisma"; +import { + createBulkOperation, + updateBulkOperationProgress, +} from "@/utils/redis/bulk-operation-progress"; + +const logger = createScopedLogger("api/webhooks/deep-clean/mark-read"); + +/** + * Webhook handler called by QStash to process mark-as-read operations. + * This is NOT a user-facing mutation - users call markCategoryAsReadAction which queues this job. + */ +export async function POST(request: Request) { + try { + const body = await request.json(); + const { emailAccountId, operationId, category, senders } = body; + + if (!emailAccountId || !operationId || !category || !senders) { + return NextResponse.json( + { error: "Missing required fields" }, + { status: 400 }, + ); + } + + logger.info("Starting mark-as-read operation", { + emailAccountId, + operationId, + category, + senderCount: senders.length, + }); + + // Validate user access + await validateUserAndAiAccess({ emailAccountId }); + + // Get email account details + const emailAccountWithAccount = await prisma.emailAccount.findUnique({ + where: { id: emailAccountId }, + select: { + email: true, + account: { + select: { + provider: true, + }, + }, + }, + }); + + const account = emailAccountWithAccount?.account; + const ownerEmail = emailAccountWithAccount?.email; + if (!account) throw new SafeError("No account found"); + if (!ownerEmail) throw new SafeError("No email found"); + + // Create email provider (works for both Gmail and Outlook) + const emailProvider = await createEmailProvider({ + emailAccountId, + provider: account.provider, + }); + + // Get all thread IDs for these senders + const allThreadIds: string[] = []; + for (const sender of senders) { + try { + const threads = await emailProvider.getThreadsFromSenderWithSubject( + sender, + 1000, + ); + allThreadIds.push(...threads.map((t) => t.id)); + } catch (error) { + logger.warn("Failed to get threads for sender", { + sender, + error, + }); + // Continue with other senders even if one fails + } + } + + if (allThreadIds.length === 0) { + logger.info("No threads found to mark as read", { + emailAccountId, + operationId, + }); + return NextResponse.json({ + ok: true, + message: "No threads found to mark as read", + operationId, + }); + } + + // Create progress tracker + await createBulkOperation({ + emailAccountId, + operationId, + operationType: "mark-read", + categoryOrSender: category, + totalItems: allThreadIds.length, + }); + + // Update status to processing + await updateBulkOperationProgress({ + emailAccountId, + operationId, + status: "processing", + }); + + let successCount = 0; + let errorCount = 0; + + // Process threads in batches + const BATCH_SIZE = 50; + for (let i = 0; i < allThreadIds.length; i += BATCH_SIZE) { + const batch = allThreadIds.slice(i, i + BATCH_SIZE); + + await Promise.all( + batch.map(async (threadId) => { + try { + await emailProvider.markReadThread(threadId, true); + successCount++; + + // Update progress every 10 threads + if (successCount % 10 === 0) { + await updateBulkOperationProgress({ + emailAccountId, + operationId, + incrementCompleted: 10, + }); + } + } catch (error) { + errorCount++; + logger.warn("Failed to mark thread as read", { + threadId, + error, + }); + } + }), + ); + + logger.info("Batch marked as read", { + operationId, + batchNumber: Math.floor(i / BATCH_SIZE) + 1, + totalBatches: Math.ceil(allThreadIds.length / BATCH_SIZE), + successCount, + errorCount, + }); + } + + // Final progress update + await updateBulkOperationProgress({ + emailAccountId, + operationId, + incrementCompleted: successCount % 10, // Update any remaining + incrementFailed: errorCount, + status: "completed", + }); + + logger.info("Mark-as-read operation completed", { + emailAccountId, + operationId, + totalThreads: allThreadIds.length, + successCount, + errorCount, + }); + + return NextResponse.json({ + ok: true, + message: `Marked ${successCount} emails as read${errorCount > 0 ? ` (${errorCount} failed)` : ""}`, + operationId, + successCount, + errorCount, + }); + } catch (error) { + logger.error("Mark-as-read operation error", { error }); + + if (error instanceof SafeError) { + return NextResponse.json({ error: error.message }, { status: 400 }); + } + + return NextResponse.json( + { error: "Internal server error" }, + { status: 500 }, + ); + } +} diff --git a/apps/web/components/DeepCleanGroupedTable.tsx b/apps/web/components/DeepCleanGroupedTable.tsx new file mode 100644 index 0000000000..5f6bfec4e6 --- /dev/null +++ b/apps/web/components/DeepCleanGroupedTable.tsx @@ -0,0 +1,750 @@ +"use client"; + +import Link from "next/link"; +import { Fragment, useMemo, useState } from "react"; +import { useQueryState } from "nuqs"; +import groupBy from "lodash/groupBy"; +import { + useReactTable, + getCoreRowModel, + getExpandedRowModel, + type ColumnDef, + flexRender, +} from "@tanstack/react-table"; +import { + ArchiveIcon, + ChevronRight, + MoreVerticalIcon, + PencilIcon, + MailOpenIcon, +} from "lucide-react"; +import { Table, TableBody, TableCell, TableRow } from "@/components/ui/table"; +import { EmailCell } from "@/components/EmailCell"; +import { useThreads } from "@/hooks/useThreads"; +import { Skeleton } from "@/components/ui/skeleton"; +import { decodeSnippet } from "@/utils/gmail/decode"; +import { formatShortDate } from "@/utils/date"; +import { cn } from "@/utils"; +import { toastSuccess, toastError } from "@/components/Toast"; +import { changeSenderCategoryAction } from "@/utils/actions/categorize"; +import { bulkSendersAction } from "@/utils/actions/deep-clean"; +import { Button } from "@/components/ui/button"; +import { Checkbox } from "@/components/ui/checkbox"; +import { getEmailUrl, getGmailSearchUrl } from "@/utils/url"; +import { MessageText } from "@/components/Typography"; +import { + DropdownMenu, + DropdownMenuContent, + DropdownMenuItem, + DropdownMenuPortal, + DropdownMenuSub, + DropdownMenuSubContent, + DropdownMenuSubTrigger, + DropdownMenuTrigger, +} from "@/components/ui/dropdown-menu"; +import type { CategoryWithRules } from "@/utils/category.server"; +import { ViewEmailButton } from "@/components/ViewEmailButton"; +import { useAccount } from "@/providers/EmailAccountProvider"; +import { useBulkOperationProgress } from "@/hooks/useDeepClean"; + +const COLUMNS = 5; + +type EmailGroup = { + address: string; + category: CategoryWithRules | null; + meta?: { width?: string }; +}; + +export function DeepCleanGroupedTable({ + emailGroups, + categories, +}: { + emailGroups: EmailGroup[]; + categories: CategoryWithRules[]; +}) { + const { emailAccountId, userEmail } = useAccount(); + const [selectedSenders, setSelectedSenders] = useState>( + new Set(), + ); + + const categoryMap = useMemo(() => { + return categories.reduce>( + (acc, category) => { + acc[category.name] = category; + return acc; + }, + {}, + ); + }, [categories]); + + const groupedEmails = useMemo(() => { + const grouped = groupBy( + emailGroups, + (group) => + categoryMap[group.category?.name || ""]?.name || "Uncategorized", + ); + + // Add empty arrays for categories without any emails + for (const category of categories) { + if (!grouped[category.name]) { + grouped[category.name] = []; + } + } + + return grouped; + }, [emailGroups, categories, categoryMap]); + + const [expanded, setExpanded] = useQueryState("expanded", { + parse: (value) => value.split(","), + serialize: (value) => value.join(","), + }); + + const columns: ColumnDef[] = useMemo( + () => [ + { + id: "checkbox", + cell: ({ row }) => ( + { + const newSelected = new Set(selectedSenders); + if (checked) { + newSelected.add(row.original.address); + } else { + newSelected.delete(row.original.address); + } + setSelectedSenders(newSelected); + }} + /> + ), + meta: { size: "40px" }, + }, + { + id: "expander", + cell: ({ row }) => { + return row.getCanExpand() ? ( + + ) : null; + }, + meta: { size: "20px" }, + }, + { + accessorKey: "address", + cell: ({ row }) => ( + +
+ +
+ + ), + }, + { + accessorKey: "preview", + cell: ({ row }) => { + return ; + }, + }, + { + accessorKey: "actions", + cell: ({ row }) => ( + + + + + + { + try { + const result = await bulkSendersAction(emailAccountId, { + senders: [row.original.address], + action: "mark-read", + category: "", + }); + + if (result?.serverError) { + toastError({ description: result.serverError }); + } else { + toastSuccess({ description: "Marking as read" }); + } + } catch (error) { + toastError({ description: "Failed to mark as read" }); + console.error("Mark as read error:", error); + } + }} + > + + Mark as Read + + { + try { + const result = await bulkSendersAction(emailAccountId, { + senders: [row.original.address], + action: "archive", + category: "", + }); + + if (result?.serverError) { + toastError({ description: result.serverError }); + } else { + toastSuccess({ description: "Archiving" }); + } + } catch (error) { + toastError({ description: "Failed to archive" }); + console.error("Archive error:", error); + } + }} + > + + Archive + + + + + Re-categorize + + + + {categories.map((category) => ( + { + const result = await changeSenderCategoryAction( + emailAccountId, + { + sender: row.original.address, + categoryId: category.id, + }, + ); + + if (result?.serverError) { + toastError({ description: result.serverError }); + } else { + toastSuccess({ description: "Category changed" }); + // Refresh the page to show updated categories + window.location.reload(); + } + }} + > + {category.name} + + ))} + + + + + + ), + meta: { size: "60px" }, + }, + ], + [categories, userEmail, emailAccountId, selectedSenders], + ); + + const table = useReactTable({ + data: emailGroups, + columns, + getRowCanExpand: () => true, + getCoreRowModel: getCoreRowModel(), + getExpandedRowModel: getExpandedRowModel(), + }); + + // Select all senders in a category + const selectAllInCategory = (senders: EmailGroup[]) => { + const newSelected = new Set(selectedSenders); + senders.forEach((sender) => newSelected.add(sender.address)); + setSelectedSenders(newSelected); + }; + + // Deselect all senders in a category + const deselectAllInCategory = (senders: EmailGroup[]) => { + const newSelected = new Set(selectedSenders); + senders.forEach((sender) => newSelected.delete(sender.address)); + setSelectedSenders(newSelected); + }; + + // Archive selected senders + const archiveSelected = async () => { + try { + const result = await bulkSendersAction(emailAccountId, { + senders: Array.from(selectedSenders), + action: "archive", + category: "", + }); + + if (result?.serverError) { + toastError({ description: result.serverError }); + } else { + toastSuccess({ + description: `Archiving ${result?.data?.count ?? selectedSenders.size} senders`, + }); + setSelectedSenders(new Set()); + } + } catch (error) { + toastError({ description: "Failed to archive senders" }); + console.error("Archive error:", error); + } + }; + + // Mark selected as read + const markSelectedAsRead = async () => { + try { + const result = await bulkSendersAction(emailAccountId, { + senders: Array.from(selectedSenders), + action: "mark-read", + category: "", + }); + + if (result?.serverError) { + toastError({ description: result.serverError }); + } else { + toastSuccess({ + description: `Marking ${result?.data?.count ?? selectedSenders.size} senders as read`, + }); + setSelectedSenders(new Set()); + } + } catch (error) { + toastError({ description: "Failed to mark senders as read" }); + console.error("Mark as read error:", error); + } + }; + + return ( + <> + {/* Bulk Actions */} + {selectedSenders.size > 0 && ( +
+ + {selectedSenders.size} sender{selectedSenders.size !== 1 ? "s" : ""}{" "} + selected + + + + +
+ )} + + + + {Object.entries(groupedEmails).map(([categoryName, senders]) => { + const isCategoryExpanded = expanded?.includes(categoryName); + const categorySelectedCount = senders.filter((sender) => + selectedSenders.has(sender.address), + ).length; + const isAllSelected = + senders.length > 0 && categorySelectedCount === senders.length; + const isPartiallySelected = + categorySelectedCount > 0 && + categorySelectedCount < senders.length; + + const onArchiveAll = async () => { + try { + const result = await bulkSendersAction(emailAccountId, { + senders: senders.map((s) => s.address), + action: "archive", + category: categoryName, + }); + + if (result?.serverError) { + toastError({ description: result.serverError }); + } else { + toastSuccess({ + description: `Archiving all ${result?.data?.count ?? senders.length} senders`, + }); + } + } catch (error) { + toastError({ description: "Failed to archive senders" }); + console.error("Archive error:", error); + } + }; + + const onMarkAllAsRead = async () => { + try { + const result = await bulkSendersAction(emailAccountId, { + senders: senders.map((s) => s.address), + action: "mark-read", + category: categoryName, + }); + + if (result?.serverError) { + toastError({ description: result.serverError }); + } else { + toastSuccess({ + description: `Marking all ${result?.data?.count ?? senders.length} senders as read`, + }); + } + } catch (error) { + toastError({ description: "Failed to mark emails as read" }); + console.error("Mark as read error:", error); + } + }; + + // const onEditCategory = () => { + // setSelectedCategoryName(categoryName); + // }; + + // const onRemoveAllFromCategory = async () => { + // const yes = confirm( + // "This will remove all emails from this category. You can re-categorize them later. Do you want to continue?", + // ); + // if (!yes) return; + // const result = await removeAllFromCategoryAction(emailAccountId, { + // categoryName, + // }); + + // if (result?.serverError) { + // toastError({ description: result.serverError }); + // } else { + // toastSuccess({ + // description: "All emails removed from category", + // }); + // } + // }; + + const category = categoryMap[categoryName]; + + if (!category) { + return null; + } + + return ( + + { + setExpanded((prev) => + isCategoryExpanded + ? (prev || []).filter((c) => c !== categoryName) + : [...(prev || []), categoryName], + ); + }} + onSelectAll={() => selectAllInCategory(senders)} + onDeselectAll={() => deselectAllInCategory(senders)} + onArchiveAll={onArchiveAll} + onMarkAllAsRead={onMarkAllAsRead} + /> + {isCategoryExpanded && ( + + )} + + ); + })} + +
+ + {/* + setSelectedCategoryName(open ? selectedCategoryName : null) + } + closeModal={() => setSelectedCategoryName(null)} + category={ + selectedCategoryName + ? categories.find((c) => c.name === selectedCategoryName) + : undefined + } + /> */} + + ); +} + +function GroupRow({ + category, + count, + isExpanded, + isAllSelected, + isPartiallySelected, + onToggle, + onSelectAll, + onDeselectAll, + onArchiveAll, + onMarkAllAsRead, +}: { + category: CategoryWithRules; + count: number; + isExpanded: boolean; + isAllSelected: boolean; + isPartiallySelected: boolean; + onToggle: () => void; + onSelectAll: () => void; + onDeselectAll: () => void; + onArchiveAll: () => void; + onMarkAllAsRead: () => void; +}) { + return ( + + { + e.stopPropagation(); + if (isAllSelected) { + onDeselectAll(); + } else { + onSelectAll(); + } + }} + > + { + if (el && "indeterminate" in el) { + // biome-ignore lint/suspicious/noExplicitAny: indeterminate is a valid property on checkbox elements + (el as any).indeterminate = isPartiallySelected; + } + }} + /> + + +
+ + {category.name} + ({count}) +
+ + + + {/* + + + + + + + Edit + + + + Remove All From Category + + + */} + + + + + + ); +} + +function SenderRows({ + table, + senders, + userEmail, +}: { + table: ReturnType>; + senders: EmailGroup[]; + userEmail: string; +}) { + if (!senders.length) { + return ( + + + This category is empty + + + ); + } + + return senders.map((sender) => { + const row = table + .getRowModel() + .rows.find((r) => r.original.address === sender.address); + if (!row) return null; + return ( + + + {row.getVisibleCells().map((cell) => ( + + {flexRender(cell.column.columnDef.cell, cell.getContext())} + + ))} + + {row.getIsExpanded() && ( + + )} + + ); + }); +} + +function ExpandedRows({ + sender, + userEmail, +}: { + sender: string; + userEmail: string; +}) { + const { provider } = useAccount(); + + // Only show emails that are currently in inbox (not archived) + // This prevents showing emails from 6 months ago that are no longer relevant + const { data, isLoading, error } = useThreads({ + fromEmail: sender, + limit: 5, + type: "inbox", // Only show inbox emails, not archived ones + }); + + if (isLoading) { + return ( + + + + + + ); + } + + if (error) { + return ( + + Error loading emails + + ); + } + + if (!data?.threads.length) { + return ( + + No emails found + + ); + } + + return ( + <> + {data.threads.map((thread) => { + const firstMessage = thread.messages[0]; + const subject = firstMessage.subject; + const date = firstMessage.date; + + return ( + + + + + + + {subject} + + + + {decodeSnippet(thread.messages[0].snippet)} + + + {formatShortDate(new Date(date))} + + + + ); + })} + + ); +} + +function OperationStatusCell({ sender }: { sender: string }) { + const { data } = useBulkOperationProgress(2000); // Poll every 2 seconds + + // Find operation for this specific sender + const operation = data?.operations.find( + (op) => op.categoryOrSender === sender, + ); + + if (!operation) return null; + + switch (operation.status) { + case "completed": { + if (operation.completedItems > 0) { + const text = + operation.operationType === "archive" + ? `Archived ${operation.completedItems} emails!` + : `Marked ${operation.completedItems} as read!`; + return {text}; + } + return Completed; + } + case "processing": { + const text = + operation.operationType === "archive" + ? `Archiving... ${operation.completedItems}/${operation.totalItems}` + : `Marking as read... ${operation.completedItems}/${operation.totalItems}`; + return {text}; + } + case "pending": + return Pending...; + case "failed": + return Failed; + default: + return null; + } +} diff --git a/apps/web/components/ProgressPanel.tsx b/apps/web/components/ProgressPanel.tsx index e4485aeb95..6459133b9f 100644 --- a/apps/web/components/ProgressPanel.tsx +++ b/apps/web/components/ProgressPanel.tsx @@ -25,7 +25,7 @@ export function ProgressPanel({ if (!totalItems) return null; return ( -
+
{ href: prefixPath(currentEmailAccountId, "/bulk-unsubscribe"), icon: MailsIcon, }, - ...(isGoogleProvider(provider) - ? [ - { - name: "Deep Clean", - href: prefixPath(currentEmailAccountId, "/clean"), - icon: BrushIcon, - }, - ] - : []), + { + name: "Deep Clean", + href: prefixPath(currentEmailAccountId, "/deep-clean"), + icon: BrushIcon, + }, + // ...(isGoogleProvider(provider) + // ? [ + // { + // name: "Bulk Clean", + // href: prefixPath(currentEmailAccountId, "/clean"), + // icon: BrushIcon, + // }, + // ] + // : []), { name: "Analytics", href: prefixPath(currentEmailAccountId, "/stats"), diff --git a/apps/web/hooks/useDeepClean.ts b/apps/web/hooks/useDeepClean.ts new file mode 100644 index 0000000000..04cb2fea68 --- /dev/null +++ b/apps/web/hooks/useDeepClean.ts @@ -0,0 +1,16 @@ +import useSWR from "swr"; +import type { DeepCleanSendersResponse } from "@/app/api/user/deep-clean/senders/route"; +import type { BulkOperationProgress } from "@/app/api/user/deep-clean/progress/route"; + +export function useDeepCleanSenders() { + return useSWR("/api/user/deep-clean/senders"); +} + +export function useBulkOperationProgress(refreshInterval?: number) { + return useSWR<{ operations: BulkOperationProgress }>( + "/api/user/deep-clean/progress", + { + refreshInterval, + }, + ); +} diff --git a/apps/web/prisma/schema.prisma b/apps/web/prisma/schema.prisma index 9129aacf84..d73404c238 100644 --- a/apps/web/prisma/schema.prisma +++ b/apps/web/prisma/schema.prisma @@ -615,6 +615,7 @@ model Newsletter { updatedAt DateTime @updatedAt email String status NewsletterStatus? + priority Int? // Sender-level priority: 10 (low), 50 (mixed), or 90 (high) // For learned patterns for rules patternAnalyzed Boolean @default(false) @@ -658,7 +659,7 @@ model EmailMessage { messageId String date DateTime // date of the email from String - fromName String? // sender's display name + fromName String? // sender's display name fromDomain String to String unsubscribeLink String? diff --git a/apps/web/utils/actions/deep-clean.ts b/apps/web/utils/actions/deep-clean.ts new file mode 100644 index 0000000000..aa002669d2 --- /dev/null +++ b/apps/web/utils/actions/deep-clean.ts @@ -0,0 +1,293 @@ +"use server"; + +import { revalidatePath } from "next/cache"; +import prisma from "@/utils/prisma"; +import { validateUserAndAiAccess } from "@/utils/user/validate"; +import { actionClient } from "@/utils/actions/safe-action"; +import { prefixPath } from "@/utils/path"; +import { + publishArchiveCategoryQueue, + publishMarkAsReadCategoryQueue, +} from "@/utils/upstash/bulk-operations"; +import { getTopSendersForDeepClean } from "@/utils/upstash/deep-clean-categorization"; +import { publishToAiCategorizeSendersQueue } from "@/utils/upstash/categorize-senders"; +import { saveCategorizationTotalItems } from "@/utils/redis/categorization-progress"; +import { + bulkCategorySchema, + bulkSendersSchema, + categorizeMoreSendersSchema, +} from "@/utils/actions/deep-clean.validation"; +import { SafeError } from "@/utils/error"; + +export const bulkCategoryAction = actionClient + .metadata({ name: "bulkCategory" }) + .schema(bulkCategorySchema) + .action( + async ({ + ctx: { emailAccountId, logger }, + parsedInput: { category, action }, + }): Promise<{ + success: boolean; + count: number; + operationId: string | null; + }> => { + await validateUserAndAiAccess({ emailAccountId }); + + logger.info("Bulk category operation", { + category, + action, + emailAccountId, + }); + + const senders = await prisma.newsletter.findMany({ + where: { + emailAccountId, + categoryId: { not: null }, + category: { name: category }, + }, + select: { email: true }, + }); + + if (senders.length === 0) { + return { + success: true, + count: 0, + operationId: null, + }; + } + + logger.info("Found senders", { + category, + senderCount: senders.length, + }); + + const operationId = `${action}-${category}-${Date.now()}`; + + switch (action) { + case "archive": { + await publishArchiveCategoryQueue({ + emailAccountId, + operationId, + category, + senders: senders.map((s) => s.email), + }); + break; + } + case "mark-read": { + await publishMarkAsReadCategoryQueue({ + emailAccountId, + operationId, + category, + senders: senders.map((s) => s.email), + }); + break; + } + default: { + logger.error("Invalid action", { action }); + throw new SafeError(`Invalid action: ${action}`); + } + } + + logger.info("Queued bulk category operation", { + emailAccountId, + operationId, + category, + senderCount: senders.length, + }); + + return { + success: true, + count: senders.length, + operationId, + }; + }, + ); + +export const bulkSendersAction = actionClient + .metadata({ name: "bulkSenders" }) + .schema(bulkSendersSchema) + .action( + async ({ + ctx: { emailAccountId, logger }, + parsedInput: { senders, action, category }, + }): Promise<{ + success: boolean; + count: number; + operationId: string | null; + }> => { + await validateUserAndAiAccess({ emailAccountId }); + + logger.info("Bulk senders operation", { + senderCount: senders.length, + action, + emailAccountId, + category, + }); + + if (senders.length === 0) { + return { + success: true, + count: 0, + operationId: null, + }; + } + + const operationId = `${action}-${Date.now()}`; + + switch (action) { + case "archive": { + await publishArchiveCategoryQueue({ + emailAccountId, + operationId, + category, + senders, + }); + break; + } + case "mark-read": { + await publishMarkAsReadCategoryQueue({ + emailAccountId, + operationId, + category, + senders, + }); + break; + } + default: { + logger.error("Invalid action", { action }); + throw new SafeError(`Invalid action: ${action}`); + } + } + + logger.info("Queued bulk senders operation", { + emailAccountId, + operationId, + senderCount: senders.length, + }); + + return { + success: true, + count: senders.length, + operationId, + }; + }, + ); + +export const categorizeMoreSendersAction = actionClient + .metadata({ name: "categorizeMoreSenders" }) + .schema(categorizeMoreSendersSchema) + .action( + async ({ ctx: { emailAccountId, logger }, parsedInput: { limit } }) => { + await validateUserAndAiAccess({ emailAccountId }); + + logger.info("Triggering categorize more senders", { + emailAccountId, + limit, + }); + + // First, try to get uncategorized senders from Newsletter table + let sendersToCategorize = await getTopSendersForDeepClean({ + emailAccountId, + limit, + }); + + logger.info("Found senders in Newsletter table", { + count: sendersToCategorize.length, + }); + + // If we have very few senders (<10), fetch more from EmailMessage table + // This handles the case where user hasn't visited bulk-unsubscribe yet + if (sendersToCategorize.length < 10) { + logger.info("Fetching additional senders from EmailMessage table"); + + // Get senders directly from EmailMessage table + const emailSenders = await prisma.emailMessage.findMany({ + where: { + emailAccountId, + sent: false, + }, + select: { + from: true, + fromName: true, + }, + distinct: ["from"], + take: limit * 2, // Get more to account for already categorized ones + orderBy: { + date: "desc", // Most recent first + }, + }); + + // Check which ones aren't already categorized + const existingCategorized = await prisma.newsletter.findMany({ + where: { + emailAccountId, + categoryId: { not: null }, + }, + select: { email: true }, + }); + + const categorizedSet = new Set(existingCategorized.map((n) => n.email)); + const newSenders = emailSenders + .filter((s) => !categorizedSet.has(s.from)) + .slice(0, limit); + + // Upsert these senders into Newsletter table so they can be tracked + await Promise.all( + newSenders.map((sender) => + prisma.newsletter.upsert({ + where: { + email_emailAccountId: { + email: sender.from, + emailAccountId, + }, + }, + update: {}, + create: { + email: sender.from, + emailAccountId, + }, + }), + ), + ); + + sendersToCategorize = newSenders.map((s) => s.from); + + logger.info("Added senders from EmailMessage", { + newCount: sendersToCategorize.length, + }); + } + + if (sendersToCategorize.length === 0) { + return { + success: true, + queuedCount: 0, + message: "No more senders to categorize", + }; + } + + // Initialize progress tracking in Redis + await saveCategorizationTotalItems({ + emailAccountId, + totalItems: sendersToCategorize.length, + }); + + // Trigger categorization using existing queue system + await publishToAiCategorizeSendersQueue({ + emailAccountId, + senders: sendersToCategorize, + }); + + logger.info("Queued more senders for categorization", { + emailAccountId, + queuedCount: sendersToCategorize.length, + }); + + revalidatePath(prefixPath(emailAccountId, "/deep-clean")); + + return { + success: true, + queuedCount: sendersToCategorize.length, + message: `Queued ${sendersToCategorize.length} senders for categorization`, + }; + }, + ); diff --git a/apps/web/utils/actions/deep-clean.validation.ts b/apps/web/utils/actions/deep-clean.validation.ts new file mode 100644 index 0000000000..3f8c8407eb --- /dev/null +++ b/apps/web/utils/actions/deep-clean.validation.ts @@ -0,0 +1,21 @@ +import { z } from "zod"; + +export const bulkCategorySchema = z.object({ + category: z.string(), + action: z.enum(["archive", "mark-read"]), +}); +export type BulkCategorySchema = z.infer; + +export const bulkSendersSchema = z.object({ + senders: z.array(z.string()), + action: z.enum(["archive", "mark-read"]), + category: z.string(), +}); +export type BulkSendersSchema = z.infer; + +export const categorizeMoreSendersSchema = z.object({ + limit: z.number().optional().default(100), +}); +export type CategorizeMoreSendersSchema = z.infer< + typeof categorizeMoreSendersSchema +>; diff --git a/apps/web/utils/ai/categorize-sender/ai-categorize-senders.ts b/apps/web/utils/ai/categorize-sender/ai-categorize-senders.ts index dd4fe02ddd..642f9246d4 100644 --- a/apps/web/utils/ai/categorize-sender/ai-categorize-senders.ts +++ b/apps/web/utils/ai/categorize-sender/ai-categorize-senders.ts @@ -6,18 +6,14 @@ import { formatCategoriesForPrompt } from "@/utils/ai/categorize-sender/format-c import { extractEmailAddress } from "@/utils/email"; import { getModel } from "@/utils/llms/model"; import { createGenerateObject } from "@/utils/llms"; - -export const REQUEST_MORE_INFORMATION_CATEGORY = "RequestMoreInformation"; -export const UNKNOWN_CATEGORY = "Unknown"; +import { + CATEGORIZE_SENDER_SYSTEM_PROMPT, + CATEGORIZATION_INSTRUCTIONS, + bulkSenderCategorizationItemSchema, +} from "@/utils/ai/categorize-sender/prompts"; const categorizeSendersSchema = z.object({ - senders: z.array( - z.object({ - rationale: z.string().describe("Keep it short."), - sender: z.string(), - category: z.string(), // not using enum, because sometimes the ai creates new categories, which throws an error. we prefer to handle this ourselves - }), - ), + senders: z.array(bulkSenderCategorizationItemSchema), }); export async function aiCategorizeSenders({ @@ -35,14 +31,11 @@ export async function aiCategorizeSenders({ { category?: string; sender: string; + priority?: "low" | "medium" | "high"; }[] > { if (senders.length === 0) return []; - const system = `You are an AI assistant specializing in email management and organization. -Your task is to categorize email accounts based on their names, email addresses, and emails they've sent us. -Provide accurate categorizations to help users efficiently manage their inbox.`; - const prompt = `Categorize the following senders: ${senders @@ -73,18 +66,10 @@ ${formatCategoriesForPrompt(categories)} -1. Analyze each sender's email address and their recent emails for categorization. -2. If the sender's category is clear, assign it. -3. Use "Unknown" if the category is unclear or multiple categories could apply. -4. Use "${REQUEST_MORE_INFORMATION_CATEGORY}" if more context is needed. - - - -- Accuracy is more important than completeness -- Only use the categories provided above -- Respond with "Unknown" if unsure -- Return your response in JSON format -`; +Analyze each sender's email address and their recent emails for categorization. + +${CATEGORIZATION_INSTRUCTIONS} +`; const modelOptions = getModel(emailAccount.user, "economy"); @@ -96,7 +81,7 @@ ${formatCategoriesForPrompt(categories)} const aiResponse = await generateObject({ ...modelOptions, - system, + system: CATEGORIZE_SENDER_SYSTEM_PROMPT, prompt, schema: categorizeSendersSchema, }); @@ -106,19 +91,7 @@ ${formatCategoriesForPrompt(categories)} senders.map((s) => s.emailAddress), ); - // filter out any senders that don't have a valid category - const results = matchedSenders.map((r) => { - if (!categories.find((c) => c.name === r.category)) { - return { - category: undefined, - sender: r.sender, - }; - } - - return r; - }); - - return results; + return matchedSenders; } // match up emails with full email @@ -143,7 +116,7 @@ function matchSendersWithFullEmail( if (!sender) return; - return { sender, category: r.category }; + return { sender, category: r.category, priority: r.priority }; }) .filter(isDefined); } diff --git a/apps/web/utils/ai/categorize-sender/ai-categorize-single-sender.ts b/apps/web/utils/ai/categorize-sender/ai-categorize-single-sender.ts index e5bac0f9e8..f1890e59d4 100644 --- a/apps/web/utils/ai/categorize-sender/ai-categorize-single-sender.ts +++ b/apps/web/utils/ai/categorize-sender/ai-categorize-single-sender.ts @@ -1,9 +1,13 @@ -import { z } from "zod"; import type { EmailAccountWithAI } from "@/utils/llms/types"; import type { Category } from "@prisma/client"; import { formatCategoriesForPrompt } from "@/utils/ai/categorize-sender/format-categories"; import { getModel } from "@/utils/llms/model"; import { createGenerateObject } from "@/utils/llms"; +import { + CATEGORIZE_SENDER_SYSTEM_PROMPT, + CATEGORIZATION_INSTRUCTIONS, + senderCategorizationSchema, +} from "@/utils/ai/categorize-sender/prompts"; export async function aiCategorizeSender({ emailAccount, @@ -16,10 +20,6 @@ export async function aiCategorizeSender({ previousEmails: { subject: string; snippet: string }[]; categories: Pick[]; }) { - const system = `You are an AI assistant specializing in email management and organization. -Your task is to categorize an email accounts based on their name, email address, and content from previous emails. -Provide an accurate categorization to help users efficiently manage their inbox.`; - const prompt = `Categorize the following email account: ${sender} @@ -38,15 +38,13 @@ ${formatCategoriesForPrompt(categories)} -1. Analyze the sender's name and email address for clues about their category. -2. Review the content of previous emails to gain more context about the account's relationship with us. -3. If the category is clear, assign it. -4. If you're not certain, respond with "Unknown". -5. If multiple categories are possible, respond with "Unknown". -6. Return your response in JSON format. +Analyze the sender's name and email address for clues about their category. +Review the content of previous emails to gain more context about the account's relationship with us. + +${CATEGORIZATION_INSTRUCTIONS} `; - const modelOptions = getModel(emailAccount.user); + const modelOptions = getModel(emailAccount.user, "economy"); const generateObject = createGenerateObject({ userEmail: emailAccount.email, @@ -56,16 +54,10 @@ ${formatCategoriesForPrompt(categories)} const aiResponse = await generateObject({ ...modelOptions, - system, + system: CATEGORIZE_SENDER_SYSTEM_PROMPT, prompt, - schema: z.object({ - rationale: z.string().describe("Keep it short. 1-2 sentences max."), - category: z.string(), - }), + schema: senderCategorizationSchema, }); - if (!categories.find((c) => c.name === aiResponse.object.category)) - return null; - return aiResponse.object; } diff --git a/apps/web/utils/ai/categorize-sender/prompts.ts b/apps/web/utils/ai/categorize-sender/prompts.ts new file mode 100644 index 0000000000..0404c76ef5 --- /dev/null +++ b/apps/web/utils/ai/categorize-sender/prompts.ts @@ -0,0 +1,44 @@ +import { z } from "zod"; + +export const CATEGORIZE_SENDER_SYSTEM_PROMPT = `You are an AI assistant specializing in email management and organization. +Your task is to categorize email accounts based on their names, email addresses, and emails they've sent us. +Provide accurate categorizations to help users efficiently manage their inbox.`; + +export const CATEGORIZATION_INSTRUCTIONS = `STRONGLY prefer using the provided categories when they fit reasonably well, even if not perfect. Only create new categories when the sender truly doesn't fit any provided category. + +When creating new categories, use broad, general terms rather than specific ones: + - "Personal" for individual people and personal correspondence + - "Marketing" covers product updates, onboarding, promotional content + - "Notifications" covers system alerts, product notifications, general notifications + - "Support" covers customer success, help desk, technical support + - "Newsletter" covers digests, updates, regular communications + +Use "Personal" for: + - Individual people and personal correspondence + - Senders that appear to be real people (not automated systems) + +Use "Unknown" only as a fallback for: + - Unclear or ambiguous senders where any categorization would be unreliable + - Senders with insufficient information to make a determination + +CRITICAL: NEVER categorize personal emails as newsletters/events/marketing. Always use "Personal" for individual correspondence. + +Assign priority levels: + - low: Newsletters, marketing, promotional content, social media notifications + - medium: Support tickets, banking notifications, general notifications, receipts + - high: Critical alerts, server monitoring, important personal communications`; + +export const senderCategorizationSchema = z.object({ + rationale: z.string().describe("Keep it short. 1 sentence."), + category: z.string(), + priority: z + .enum(["low", "medium", "high"]) + .describe( + "Priority level: low (newsletters/marketing), medium (notifications/support), high (critical alerts/personal).", + ), +}); + +export const bulkSenderCategorizationItemSchema = + senderCategorizationSchema.extend({ + sender: z.string(), + }); diff --git a/apps/web/utils/categorize/senders/categorize.ts b/apps/web/utils/categorize/senders/categorize.ts index e1af709bd3..fd8103c5b2 100644 --- a/apps/web/utils/categorize/senders/categorize.ts +++ b/apps/web/utils/categorize/senders/categorize.ts @@ -5,12 +5,12 @@ import { isNewsletterSender } from "@/utils/ai/group/find-newsletters"; import { isReceiptSender } from "@/utils/ai/group/find-receipts"; import { aiCategorizeSender } from "@/utils/ai/categorize-sender/ai-categorize-single-sender"; import type { Category } from "@prisma/client"; -import { getUserCategories } from "@/utils/category.server"; import type { EmailAccountWithAI } from "@/utils/llms/types"; import { createScopedLogger } from "@/utils/logger"; import { extractEmailAddress } from "@/utils/email"; -import { SafeError } from "@/utils/error"; import type { EmailProvider } from "@/utils/email/types"; +import { priorityToNumber, type PriorityLevel } from "@/utils/priority"; +import { getUserCategories } from "@/utils/category.server"; const logger = createScopedLogger("categorize/senders"); @@ -18,12 +18,10 @@ export async function categorizeSender( senderAddress: string, emailAccount: EmailAccountWithAI, provider: EmailProvider, - userCategories?: Pick[], ) { - const categories = - userCategories || - (await getUserCategories({ emailAccountId: emailAccount.id })); - if (categories.length === 0) return { categoryId: undefined }; + const categories = await getUserCategories({ + emailAccountId: emailAccount.id, + }); const previousEmails = await provider.getThreadsFromSenderWithSubject( senderAddress, @@ -43,6 +41,7 @@ export async function categorizeSender( categories, categoryName: aiResult.category, emailAccountId: emailAccount.id, + priority: aiResult.priority, }); return { categoryId: newsletter.categoryId }; @@ -61,37 +60,61 @@ export async function updateSenderCategory({ sender, categories, categoryName, + priority, }: { emailAccountId: string; sender: string; categories: Pick[]; categoryName: string; + priority?: PriorityLevel; }) { + // Check if category exists in the provided categories array first let category = categories.find((c) => c.name === categoryName); let newCategory: Category | undefined; if (!category) { - // create category - newCategory = await prisma.category.create({ - data: { + // Category doesn't exist in provided categories, upsert it in the database + const defaultCategoryMatch = Object.values(defaultCategory).find( + (c) => c.name === categoryName, + ); + + const upsertedCategory = await prisma.category.upsert({ + where: { + name_emailAccountId: { + name: categoryName, + emailAccountId, + }, + }, + update: {}, // No updates needed since we're just ensuring it exists + create: { name: categoryName, + description: defaultCategoryMatch?.description, emailAccountId, - // color: getRandomColor(), }, }); - category = newCategory; + + category = upsertedCategory; + // Check if this was a newly created category + if (!categories.some((c) => c.id === upsertedCategory.id)) { + newCategory = upsertedCategory; + } } // save category + const priorityNumber = priority ? priorityToNumber(priority) : null; const newsletter = await prisma.newsletter.upsert({ where: { email_emailAccountId: { email: sender, emailAccountId }, }, - update: { categoryId: category.id }, + update: { + categoryId: category.id, + priority: priorityNumber, + }, create: { email: sender, emailAccountId, categoryId: category.id, + priority: priorityNumber, }, }); @@ -138,14 +161,17 @@ function preCategorizeSendersWithStaticRules( }); } -export async function getCategories({ - emailAccountId, -}: { - emailAccountId: string; -}) { - const categories = await getUserCategories({ emailAccountId }); - if (categories.length === 0) throw new SafeError("No categories found"); - return { categories }; +export async function getCategories() { + // Use default categories instead of requiring user-created categories + const defaultCategories = Object.values(defaultCategory) + .filter((category) => category.enabled) + .map((category) => ({ + id: category.name, // Use name as id for default categories + name: category.name, + description: category.description, + })); + + return { categories: defaultCategories }; } export async function categorizeWithAi({ diff --git a/apps/web/utils/email.test.ts b/apps/web/utils/email.test.ts index 33f933d4df..ba03a0c9c2 100644 --- a/apps/web/utils/email.test.ts +++ b/apps/web/utils/email.test.ts @@ -30,6 +30,70 @@ describe("email utils", () => { it("handles empty input", () => { expect(extractNameFromEmail("")).toBe(""); }); + + it("removes surrounding double quotes from names", () => { + expect(extractNameFromEmail('"vercel[bot]" ')).toBe( + "vercel[bot]", + ); + }); + + it("removes surrounding single quotes from names", () => { + expect(extractNameFromEmail("'github[bot]' ")).toBe( + "github[bot]", + ); + }); + + it("removes quotes from plain email format", () => { + expect(extractNameFromEmail('"vercel[bot]"')).toBe("vercel[bot]"); + expect(extractNameFromEmail("'github[bot]'")).toBe("github[bot]"); + }); + + it("removes quotes from bracketed email format", () => { + expect(extractNameFromEmail('<"vercel[bot]"@vercel.com>')).toBe( + "vercel[bot]", + ); + expect(extractNameFromEmail("<'github[bot]'@github.com>")).toBe( + "github[bot]", + ); + }); + + it("handles mixed quote types", () => { + expect( + extractNameFromEmail( + "\"name with 'inner' quotes\" ", + ), + ).toBe("name with 'inner' quotes"); + expect( + extractNameFromEmail( + "'name with \"inner\" quotes' ", + ), + ).toBe('name with "inner" quotes'); + }); + + it("handles names with brackets and quotes", () => { + expect(extractNameFromEmail('"[bot] system" ')).toBe( + "[bot] system", + ); + expect( + extractNameFromEmail("'[automated] service' "), + ).toBe("[automated] service"); + }); + + it("handles edge cases with quotes", () => { + expect(extractNameFromEmail('""')).toBe(""); + expect(extractNameFromEmail("''")).toBe(""); + expect(extractNameFromEmail('" "')).toBe(" "); + expect(extractNameFromEmail("' '")).toBe(" "); + }); + + it("handles multiple quotes correctly", () => { + expect(extractNameFromEmail('""quoted name"" ')).toBe( + '"quoted name"', + ); + expect(extractNameFromEmail("''quoted name'' ")).toBe( + "'quoted name'", + ); + }); }); describe("extractEmailAddress", () => { diff --git a/apps/web/utils/email.ts b/apps/web/utils/email.ts index 4fc3ddc332..bf3e6ea83b 100644 --- a/apps/web/utils/email.ts +++ b/apps/web/utils/email.ts @@ -6,13 +6,41 @@ const emailSchema = z.string().email(); // Converts "John Doe " to "John Doe" // Converts "" to "john.doe@gmail" // Converts "john.doe@gmail" to "john.doe@gmail" +// Removes surrounding quotes from names like "vercel[bot]" -> vercel[bot] export function extractNameFromEmail(email: string) { if (!email) return ""; + + // Handle format: "Name" or Name const firstPart = email.split("<")[0]?.trim(); - if (firstPart) return firstPart; + if (firstPart) { + const sanitizedName = removeSurroundingQuotes(firstPart); + // Only return the name if it's not empty after sanitization + if (sanitizedName) { + return sanitizedName; + } + } + + // Handle format: or <"quoted-name"@domain.com> const secondPart = email.split("<")?.[1]?.trim(); - if (secondPart) return secondPart.split(">")[0]; - return email; + if (secondPart) { + const emailPart = secondPart.split(">")[0]; + + // Check if the email part starts with a quoted name + // Pattern: "quoted-name"@domain.com or 'quoted-name'@domain.com + const quotedNameMatch = emailPart.match(/^["']([^"']+)["']@/); + if (quotedNameMatch) { + return quotedNameMatch[1]; + } + + // If no quoted name pattern, return the email part as-is + return emailPart; + } + + return removeSurroundingQuotes(email); +} + +function removeSurroundingQuotes(str: string): string { + return str.replace(/^["']|["']$/g, ""); } // Converts "John Doe " to "john.doe@gmail" diff --git a/apps/web/utils/priority.ts b/apps/web/utils/priority.ts new file mode 100644 index 0000000000..726e5b32da --- /dev/null +++ b/apps/web/utils/priority.ts @@ -0,0 +1,18 @@ +export type PriorityLevel = "low" | "medium" | "high"; + +export const PRIORITY_MAP: Record = { + low: 10, + medium: 50, + high: 90, +} as const; + +export function priorityToNumber(priority: PriorityLevel): number { + return PRIORITY_MAP[priority]; +} + +export function numberToPriority(priority: number): PriorityLevel | null { + const entry = Object.entries(PRIORITY_MAP).find( + ([, value]) => value === priority, + ); + return entry ? (entry[0] as PriorityLevel) : null; +} diff --git a/apps/web/utils/redis/bulk-operation-progress.ts b/apps/web/utils/redis/bulk-operation-progress.ts new file mode 100644 index 0000000000..226bae59a1 --- /dev/null +++ b/apps/web/utils/redis/bulk-operation-progress.ts @@ -0,0 +1,139 @@ +import { z } from "zod"; +import { redis } from "@/utils/redis"; + +export type BulkOperationType = "archive" | "mark-read"; + +const bulkOperationProgressSchema = z.object({ + operationType: z.enum(["archive", "mark-read"]), + categoryOrSender: z.string(), // category name or sender email + totalItems: z.number().int().min(0), + completedItems: z.number().int().min(0), + failedItems: z.number().int().min(0), + status: z.enum(["pending", "processing", "completed", "failed"]), + createdAt: z.string(), + updatedAt: z.string(), +}); + +export type RedisBulkOperationProgress = z.infer< + typeof bulkOperationProgressSchema +>; + +function getKey({ + emailAccountId, + operationId, +}: { + emailAccountId: string; + operationId: string; +}) { + return `bulk-operation-progress:${emailAccountId}:${operationId}`; +} + +export async function getBulkOperationProgress({ + emailAccountId, + operationId, +}: { + emailAccountId: string; + operationId: string; +}) { + const key = getKey({ emailAccountId, operationId }); + const progress = await redis.get(key); + return progress; +} + +export async function createBulkOperation({ + emailAccountId, + operationId, + operationType, + categoryOrSender, + totalItems, +}: { + emailAccountId: string; + operationId: string; + operationType: BulkOperationType; + categoryOrSender: string; + totalItems: number; +}) { + const key = getKey({ emailAccountId, operationId }); + const now = new Date().toISOString(); + + const progress: RedisBulkOperationProgress = { + operationType, + categoryOrSender, + totalItems, + completedItems: 0, + failedItems: 0, + status: "pending", + createdAt: now, + updatedAt: now, + }; + + // Store progress for 10 minutes + await redis.set(key, progress, { ex: 10 * 60 }); + return progress; +} + +export async function updateBulkOperationProgress({ + emailAccountId, + operationId, + incrementCompleted = 0, + incrementFailed = 0, + status, +}: { + emailAccountId: string; + operationId: string; + incrementCompleted?: number; + incrementFailed?: number; + status?: RedisBulkOperationProgress["status"]; +}) { + const existingProgress = await getBulkOperationProgress({ + emailAccountId, + operationId, + }); + if (!existingProgress) return null; + + const key = getKey({ emailAccountId, operationId }); + const updatedProgress: RedisBulkOperationProgress = { + ...existingProgress, + completedItems: existingProgress.completedItems + incrementCompleted, + failedItems: existingProgress.failedItems + incrementFailed, + status: status || existingProgress.status, + updatedAt: new Date().toISOString(), + }; + + // Auto-complete if all items are processed + if ( + updatedProgress.completedItems + updatedProgress.failedItems >= + updatedProgress.totalItems + ) { + updatedProgress.status = "completed"; + } + + // Store progress for 10 minutes + await redis.set(key, updatedProgress, { ex: 10 * 60 }); + return updatedProgress; +} + +export async function getAllBulkOperations({ + emailAccountId, +}: { + emailAccountId: string; +}): Promise> { + const pattern = `bulk-operation-progress:${emailAccountId}:*`; + const keys = await redis.keys(pattern); + + if (keys.length === 0) return []; + + const operations = await Promise.all( + keys.map(async (key) => { + const operationId = key.split(":").pop()!; + const progress = await redis.get(key); + if (!progress) return null; + return { ...progress, operationId }; + }), + ); + + return operations.filter( + (op): op is RedisBulkOperationProgress & { operationId: string } => + op !== null, + ); +} diff --git a/apps/web/utils/upstash/bulk-operations.ts b/apps/web/utils/upstash/bulk-operations.ts new file mode 100644 index 0000000000..a0227da914 --- /dev/null +++ b/apps/web/utils/upstash/bulk-operations.ts @@ -0,0 +1,99 @@ +import { publishToQstashQueue } from "@/utils/upstash"; +import { env } from "@/env"; +import { createScopedLogger } from "@/utils/logger"; + +const logger = createScopedLogger("upstash/bulk-operations"); + +const BULK_OPERATIONS_PREFIX = "bulk-operations"; + +const getBulkOperationQueueName = ({ + emailAccountId, + operationType, +}: { + emailAccountId: string; + operationType: "archive" | "mark-read"; +}) => `${BULK_OPERATIONS_PREFIX}-${operationType}-${emailAccountId}`; + +/** + * Publishes an archive category operation to the queue + */ +export async function publishArchiveCategoryQueue({ + emailAccountId, + operationId, + category, + senders, +}: { + emailAccountId: string; + operationId: string; + category: string; + senders: string[]; +}) { + const url = `${env.WEBHOOK_URL || env.NEXT_PUBLIC_BASE_URL}/api/webhooks/deep-clean/archive`; + + const queueName = getBulkOperationQueueName({ + emailAccountId, + operationType: "archive", + }); + + logger.info("Publishing archive category operation", { + emailAccountId, + operationId, + category, + senderCount: senders.length, + queueName, + }); + + await publishToQstashQueue({ + queueName, + parallelism: 1, // Process one category at a time per user + url, + body: { + emailAccountId, + operationId, + category, + senders, + }, + }); +} + +/** + * Publishes a mark-as-read category operation to the queue + */ +export async function publishMarkAsReadCategoryQueue({ + emailAccountId, + operationId, + category, + senders, +}: { + emailAccountId: string; + operationId: string; + category: string; + senders: string[]; +}) { + const url = `${env.WEBHOOK_URL || env.NEXT_PUBLIC_BASE_URL}/api/webhooks/deep-clean/mark-read`; + + const queueName = getBulkOperationQueueName({ + emailAccountId, + operationType: "mark-read", + }); + + logger.info("Publishing mark-as-read category operation", { + emailAccountId, + operationId, + category, + senderCount: senders.length, + queueName, + }); + + await publishToQstashQueue({ + queueName, + parallelism: 1, // Process one category at a time per user + url, + body: { + emailAccountId, + operationId, + category, + senders, + }, + }); +} diff --git a/apps/web/utils/upstash/deep-clean-categorization.ts b/apps/web/utils/upstash/deep-clean-categorization.ts new file mode 100644 index 0000000000..fe8c7434b5 --- /dev/null +++ b/apps/web/utils/upstash/deep-clean-categorization.ts @@ -0,0 +1,108 @@ +import { publishToQstashQueue } from "@/utils/upstash"; +import { env } from "@/env"; +import { createScopedLogger } from "@/utils/logger"; +import prisma from "@/utils/prisma"; + +const logger = createScopedLogger("upstash/deep-clean"); + +const DEEP_CLEAN_CATEGORIZATION_PREFIX = "deep-clean-categorization"; + +const getDeepCleanQueueName = ({ + emailAccountId, +}: { + emailAccountId: string; +}) => `${DEEP_CLEAN_CATEGORIZATION_PREFIX}-${emailAccountId}`; + +/** + * Triggers DeepClean categorization for a user's top senders + * This should be called 2-5 minutes after user signup + */ +export async function triggerDeepCleanCategorization({ + emailAccountId, +}: { + emailAccountId: string; +}) { + const url = `${env.WEBHOOK_URL || env.NEXT_PUBLIC_BASE_URL}/api/user/deep-clean/categorize`; + + const queueName = getDeepCleanQueueName({ emailAccountId }); + + logger.info("Triggering DeepClean categorization", { + emailAccountId, + url, + queueName, + }); + + await publishToQstashQueue({ + queueName, + parallelism: 1, // Only one job per user + url, + body: { + emailAccountId, + }, + // Note: delay param not currently supported, consider using QStash schedule API if needed + }); +} + +/** + * Gets the top senders by email count for DeepClean categorization + * Returns senders that haven't been categorized yet + * Filters out senders with very few emails (likely personal contacts) + */ +export async function getTopSendersForDeepClean({ + emailAccountId, + limit = 100, + minEmailCount = 3, +}: { + emailAccountId: string; + limit?: number; + minEmailCount?: number; // Minimum number of emails from a sender to categorize them +}) { + // Get top senders by email count from EmailMessage table + const topSenders = await prisma.emailMessage.groupBy({ + by: ["from"], + where: { + emailAccountId, + inbox: true, // Only count inbox emails + }, + _count: { + from: true, + }, + orderBy: { + _count: { + from: "desc", + }, + }, + take: limit * 3, // Get more to account for already categorized ones and filtered low-count senders + }); + + // Filter out senders that are already categorized + const categorizedSenders = await prisma.newsletter.findMany({ + where: { + emailAccountId, + categoryId: { not: null }, + }, + select: { email: true }, + }); + + const categorizedEmails = new Set(categorizedSenders.map((s) => s.email)); + + // Filter for senders with enough emails and not yet categorized + // This helps avoid categorizing personal contacts with just 1-2 emails + const uncategorizedSenders = topSenders + .filter((sender) => sender._count.from >= minEmailCount) // Only include senders with minimum email count + .map((sender) => sender.from) + .filter((email) => !categorizedEmails.has(email)) + .slice(0, limit); + + logger.info("Found top senders for DeepClean", { + emailAccountId, + totalFound: topSenders.length, + uncategorizedCount: uncategorizedSenders.length, + filteredByMinCount: topSenders.filter((s) => s._count.from < minEmailCount) + .length, + requestedLimit: limit, + minEmailCount, + }); + + return uncategorizedSenders; +}