Learned patterns from labels ( cold emails only )

apps/web/app/api/google/webhook/process-history-item.test.ts

@@ -3,7 +3,8 @@ import {
   shouldRunColdEmailBlocker,
   processHistoryItem,
 } from "./process-history-item";
-import { ColdEmailSetting } from "@prisma/client";
+import { HistoryEventType } from "./types";
+import { ColdEmailSetting, ColdEmailStatus } from "@prisma/client";
 import type { gmail_v1 } from "@googleapis/gmail";
 import { isAssistantEmail } from "@/utils/assistant/is-assistant-email";
 import { runColdEmailBlockerWithProvider } from "@/utils/cold-email/is-cold-email";
@@ -16,6 +17,10 @@ import { runRules } from "@/utils/ai/choose-rule/run-rules";
 import { processAssistantEmail } from "@/utils/assistant/process-assistant-email";
 import { getEmailAccount } from "@/__tests__/helpers";
 import { enqueueDigestItem } from "@/utils/digest/index";
+import prisma from "@/utils/__mocks__/prisma";
+import { saveLearnedPatterns } from "@/utils/rule/learned-patterns";
+import { createEmailProvider } from "@/utils/email/provider";
+import { inboxZeroLabels } from "@/utils/label";
 
 vi.mock("server-only", () => ({}));
 vi.mock("next/server", () => ({
@@ -91,6 +96,26 @@ vi.mock("@/utils/email/provider", () => ({
   }),
 }));
 
+vi.mock("@/utils/gmail/label", async () => {
+  const actual = await vi.importActual("@/utils/gmail/label");
+  return {
+    ...actual,
+    getLabelById: vi.fn().mockImplementation(async ({ id }: { id: string }) => {
+      const labelMap: Record<string, { name: string }> = {
+        "label-1": { name: inboxZeroLabels.cold_email.name },
+        "label-2": { name: "Newsletter" },
+        "label-3": { name: "Marketing" },
+        "label-4": { name: "To Reply" },
+      };
+      return labelMap[id] || { name: "Unknown Label" };
+    }),
+  };
+});
+
+vi.mock("@/utils/rule/learned-patterns", () => ({
+  saveLearnedPatterns: vi.fn().mockResolvedValue(undefined),
+}));
+
 describe("processHistoryItem", () => {
   beforeEach(() => {
     vi.clearAllMocks();
@@ -99,9 +124,34 @@ describe("processHistoryItem", () => {
   const createHistoryItem = (
     messageId = "123",
     threadId = "thread-123",
-  ): gmail_v1.Schema$HistoryMessageAdded => ({
-    message: { id: messageId, threadId },
-  });
+    type: HistoryEventType = HistoryEventType.MESSAGE_ADDED,
+    labelIds?: string[],
+  ) => {
+    const baseItem = { message: { id: messageId, threadId } };
+
+    if (type === HistoryEventType.LABEL_REMOVED) {
+      return {
+        type,
+        item: {
+          ...baseItem,
+          labelIds: labelIds || [],
+        } as gmail_v1.Schema$HistoryLabelRemoved,
+      };
+    } else if (type === HistoryEventType.LABEL_ADDED) {
+      return {
+        type,
+        item: {
+          ...baseItem,
+          labelIds: labelIds || [],
+        } as gmail_v1.Schema$HistoryLabelAdded,
+      };
+    } else {
+      return {
+        type,
+        item: baseItem as gmail_v1.Schema$HistoryMessageAdded,
+      };
+    }
+  };
 
   const defaultOptions = {
     gmail: {} as any,
@@ -160,7 +210,6 @@ describe("processHistoryItem", () => {
   });
 
   it("should skip if message is outbound", async () => {
-    const { createEmailProvider } = await import("@/utils/email/provider");
     vi.mocked(createEmailProvider).mockResolvedValueOnce({
       getMessage: vi.fn().mockResolvedValue({
         id: "123",

apps/web/app/api/google/webhook/process-history-item.ts

@@ -25,11 +25,17 @@ import type { EmailAccountWithAI } from "@/utils/llms/types";
 import { formatError } from "@/utils/error";
 import { createEmailProvider } from "@/utils/email/provider";
 import { enqueueDigestItem } from "@/utils/digest/index";
+import { HistoryEventType } from "@/app/api/google/webhook/types";
+import { handleLabelRemovedEvent } from "@/app/api/google/webhook/process-label-removed-event";
 
 export async function processHistoryItem(
-  {
-    message,
-  }: gmail_v1.Schema$HistoryMessageAdded | gmail_v1.Schema$HistoryLabelAdded,
+  historyItem: {
+    type: HistoryEventType;
+    item:
+      | gmail_v1.Schema$HistoryMessageAdded
+      | gmail_v1.Schema$HistoryLabelAdded
+      | gmail_v1.Schema$HistoryLabelRemoved;
+  },
   {
     gmail,
     emailAccount,
@@ -39,8 +45,10 @@ export async function processHistoryItem(
     rules,
   }: ProcessHistoryOptions,
 ) {
-  const messageId = message?.id;
-  const threadId = message?.threadId;
+  const { type, item } = historyItem;
+  const messageId = item.message?.id;
+  const threadId = item.message?.threadId;
+
   const emailAccountId = emailAccount.id;
   const userEmail = emailAccount.email;
 
@@ -52,6 +60,23 @@ export async function processHistoryItem(
     threadId,
   };
 
+  const provider = await createEmailProvider({
+    emailAccountId,
+    provider: "google",
+  });
+
+  if (type === HistoryEventType.LABEL_REMOVED) {
+    logger.info("Processing label removed event for learning", loggerOptions);
+    return handleLabelRemovedEvent(item, {
+      gmail,
+      emailAccount,
+      provider,
+    });
+  } else if (type === HistoryEventType.LABEL_ADDED) {
+    logger.info("Processing label added event for learning", loggerOptions);
+    return;
+  }
+
   const isFree = await markMessageAsProcessing({ userEmail, messageId });
 
   if (!isFree) {
@@ -61,14 +86,9 @@ export async function processHistoryItem(
 
   logger.info("Getting message", loggerOptions);
 
-  const emailProvider = await createEmailProvider({
-    emailAccountId,
-    provider: "google",
-  });
-
   try {
     const [parsedMessage, hasExistingRule] = await Promise.all([
-      emailProvider.getMessage(messageId),
+      provider.getMessage(messageId),
       prisma.executedRule.findUnique({
         where: {
           unique_emailAccount_thread_message: {
@@ -97,11 +117,6 @@ export async function processHistoryItem(
       emailToCheck: parsedMessage.headers.to,
     });
 
-    const provider = await createEmailProvider({
-      emailAccountId,
-      provider: "google",
-    });
-
     if (isForAssistant) {
       logger.info("Passing through assistant email.", loggerOptions);
       return processAssistantEmail({

apps/web/app/api/google/webhook/process-history.ts

@@ -8,7 +8,10 @@ import { ColdEmailSetting } from "@prisma/client";
 import { captureException } from "@/utils/error";
 import { unwatchEmails } from "@/app/api/watch/controller";
 import { createEmailProvider } from "@/utils/email/provider";
-import type { ProcessHistoryOptions } from "@/app/api/google/webhook/types";
+import {
+  HistoryEventType,
+  type ProcessHistoryOptions,
+} from "@/app/api/google/webhook/types";
 import { processHistoryItem } from "@/app/api/google/webhook/process-history-item";
 import { logger } from "@/app/api/google/webhook/logger";
 import { getHistory } from "@/utils/gmail/history";
@@ -162,7 +165,7 @@ export async function processHistoryForUser(
       // NOTE this can cause problems if we're way behind
       // NOTE this doesn't include startHistoryId in the results
       startHistoryId,
-      historyTypes: ["messageAdded", "labelAdded"],
+      historyTypes: ["messageAdded", "labelAdded", "labelRemoved"],
       maxResults: 500,
     });
 
@@ -232,26 +235,43 @@ async function processHistory(options: ProcessHistoryOptions) {
     const historyMessages = [
       ...(h.messagesAdded || []),
       ...(h.labelsAdded || []),
+      ...(h.labelsRemoved || []),
     ];
 
     if (!historyMessages.length) continue;
 
-    const inboxMessages = historyMessages.filter(isInboxOrSentMessage);
-    const uniqueMessages = uniqBy(inboxMessages, (m) => m.message?.id);
+    const allEvents = [
+      ...(h.messagesAdded || [])
+        .filter(isInboxOrSentMessage)
+        .map((m) => ({ type: HistoryEventType.MESSAGE_ADDED, item: m })),
+      ...(h.labelsAdded || []).map((m) => ({
+        type: HistoryEventType.LABEL_ADDED,
+        item: m,
+      })),
+      ...(h.labelsRemoved || []).map((m) => ({
+        type: HistoryEventType.LABEL_REMOVED,
+        item: m,
+      })),
+    ];
+
+    const uniqueEvents = uniqBy(
+      allEvents,
+      (e) => `${e.type}:${e.item.message?.id}`,
+    );
 
-    for (const m of uniqueMessages) {
+    for (const event of uniqueEvents) {
       try {
-        await processHistoryItem(m, options);
+        await processHistoryItem(event, options);
       } catch (error) {
         captureException(
           error,
-          { extra: { userEmail, messageId: m.message?.id } },
+          { extra: { userEmail, messageId: event.item.message?.id } },
           userEmail,
         );
         logger.error("Error processing history item", {
           userEmail,
-          messageId: m.message?.id,
-          threadId: m.message?.threadId,
+          messageId: event.item.message?.id,
+          threadId: event.item.message?.threadId,
           error:
             error instanceof Error
               ? {