Centralised watch emails

apps/web/app/api/watch/all/route.ts

@@ -1,139 +1,23 @@
 import { NextResponse } from "next/server";
-import prisma from "@/utils/prisma";
-import { watchEmails } from "../controller";
-import { createEmailProvider } from "@/utils/email/provider";
 import { hasCronSecret, hasPostCronSecret } from "@/utils/cron";
 import { withError } from "@/utils/middleware";
 import { captureException } from "@/utils/error";
-import { hasAiAccess } from "@/utils/premium";
 import { createScopedLogger } from "@/utils/logger";
+import { ensureEmailAccountsWatched } from "@/utils/email/watch-manager";
 
 const logger = createScopedLogger("api/watch/all");
 
 export const dynamic = "force-dynamic";
 export const maxDuration = 300;
 
 async function watchAllEmails() {
-  const emailAccounts = await prisma.emailAccount.findMany({
-    where: {
-      user: {
-        premium: {
-          OR: [
-            { lemonSqueezyRenewsAt: { gt: new Date() } },
-            { stripeSubscriptionStatus: { in: ["active", "trialing"] } },
-          ],
-        },
-      },
-    },
-    select: {
-      id: true,
-      email: true,
-      watchEmailsExpirationDate: true,
-      watchEmailsSubscriptionId: true,
-      account: {
-        select: {
-          provider: true,
-          access_token: true,
-          refresh_token: true,
-          expires_at: true,
-        },
-      },
-      user: {
-        select: {
-          aiApiKey: true,
-          premium: { select: { tier: true } },
-        },
-      },
-    },
-    orderBy: {
-      watchEmailsExpirationDate: { sort: "asc", nulls: "first" },
-    },
-  });
-
-  logger.info("Watching email accounts", { count: emailAccounts.length });
-
-  for (const emailAccount of emailAccounts) {
-    try {
-      logger.info("Watching emails for account", {
-        emailAccountId: emailAccount.id,
-        email: emailAccount.email,
-        provider: emailAccount.account.provider,
-      });
-
-      const userHasAiAccess = hasAiAccess(
-        emailAccount.user.premium?.tier || null,
-        emailAccount.user.aiApiKey,
-      );
-
-      if (!userHasAiAccess) {
-        logger.info("User does not have access to AI or cold email", {
-          email: emailAccount.email,
-        });
-        if (
-          emailAccount.watchEmailsExpirationDate &&
-          new Date(emailAccount.watchEmailsExpirationDate) < new Date()
-        ) {
-          await prisma.emailAccount.update({
-            where: { email: emailAccount.email },
-            data: {
-              watchEmailsExpirationDate: null,
-              watchEmailsSubscriptionId: null,
-            },
-          });
-        }
-
-        continue;
-      }
-
-      if (
-        !emailAccount.account?.access_token ||
-        !emailAccount.account?.refresh_token
-      ) {
-        logger.info("User has no access token or refresh token", {
-          email: emailAccount.email,
-        });
-        continue;
-      }
-
-      const provider = await createEmailProvider({
-        emailAccountId: emailAccount.id,
-        provider: emailAccount.account.provider,
-      });
-
-      const result = await watchEmails({
-        emailAccountId: emailAccount.id,
-        provider,
-      });
-
-      if (!result.success) {
-        logger.error("Failed to watch emails for account", {
-          emailAccountId: emailAccount.id,
-          email: emailAccount.email,
-          error: result.error,
-        });
-      }
-    } catch (error) {
-      if (error instanceof Error) {
-        const warn = [
-          "invalid_grant",
-          "Mail service not enabled",
-          "Insufficient Permission",
-        ];
-
-        if (warn.some((w) => error.message.includes(w))) {
-          logger.warn("Not watching emails for user", {
-            email: emailAccount.email,
-            error,
-          });
-          continue;
-        }
-      }
-
-      logger.error("Error for user", { email: emailAccount.email, error });
-    }
+  try {
+    const results = await ensureEmailAccountsWatched({ userIds: null });
+    return NextResponse.json({ success: true, results });
+  } catch (error) {
+    logger.error("Failed to watch all emails", { error });
+    throw error;
   }
-
-  return NextResponse.json({ success: true });
 }
 
 export const GET = withError(async (request) => {

apps/web/app/api/watch/route.ts

@@ -1,97 +1,24 @@
 import { NextResponse } from "next/server";
 import { withAuth } from "@/utils/middleware";
-import { createScopedLogger } from "@/utils/logger";
 import prisma from "@/utils/prisma";
-import { watchEmails } from "./controller";
-import { createEmailProvider } from "@/utils/email/provider";
+import { ensureEmailAccountsWatched } from "@/utils/email/watch-manager";
 
 export const dynamic = "force-dynamic";
 
-const logger = createScopedLogger("api/watch");
-
 export const GET = withAuth(async (request) => {
   const userId = request.auth.userId;
-  const results = [];
-
-  const emailAccounts = await prisma.emailAccount.findMany({
+  const emailAccountCount = await prisma.emailAccount.count({
     where: { userId },
-    select: {
-      id: true,
-      account: {
-        select: {
-          provider: true,
-          access_token: true,
-          refresh_token: true,
-          expires_at: true,
-        },
-      },
-    },
   });
 
-  if (emailAccounts.length === 0) {
+  if (emailAccountCount === 0) {
     return NextResponse.json(
       { message: "No email accounts found for this user." },
       { status: 404 },
     );
   }
 
-  for (const { id: emailAccountId, account } of emailAccounts) {
-    try {
-      // Check for missing tokens for Microsoft accounts
-      if (!account.access_token || !account.refresh_token) {
-        logger.warn("Missing tokens for account", { emailAccountId });
-        results.push({
-          emailAccountId,
-          status: "error",
-          message: "Missing authentication tokens.",
-        });
-        continue;
-      }
-
-      // Create email provider for this account
-      const provider = await createEmailProvider({
-        emailAccountId,
-        provider: account.provider,
-      });
-
-      const result = await watchEmails({
-        emailAccountId,
-        provider,
-      });
-
-      if (result.success) {
-        results.push({
-          emailAccountId,
-          status: "success",
-          expirationDate: result.expirationDate,
-        });
-      } else {
-        logger.error("Error watching inbox for account", {
-          emailAccountId,
-          provider: account.provider,
-          error: result.error,
-        });
-        results.push({
-          emailAccountId,
-          status: "error",
-          message: "Failed to set up watch for this account.",
-          errorDetails: result.error,
-        });
-      }
-    } catch (error) {
-      logger.error("Exception while watching inbox for account", {
-        emailAccountId,
-        error,
-      });
-      results.push({
-        emailAccountId,
-        status: "error",
-        message:
-          "An unexpected error occurred while setting up watch for this account.",
-        errorDetails: error instanceof Error ? error.message : String(error),
-      });
-    }
-  }
+  const results = await ensureEmailAccountsWatched({ userIds: [userId] });
 
   return NextResponse.json({ results });
 });

apps/web/ee/billing/stripe/sync-stripe.ts

@@ -1,10 +1,12 @@
+import { after } from "next/server";
 import sumBy from "lodash/sumBy";
 import prisma from "@/utils/prisma";
 import { createScopedLogger } from "@/utils/logger";
 import { getStripe } from "@/ee/billing/stripe";
 import { getStripeSubscriptionTier } from "@/app/(app)/premium/config";
 import { handleLoopsEvents } from "@/ee/billing/stripe/loops-events";
 import { updateAccountSeatsForPremium } from "@/utils/premium/server";
+import { ensureEmailAccountsWatched } from "@/utils/email/watch-manager";
 import type { Prisma } from "@prisma/client";
 
 const logger = createScopedLogger("stripe/syncStripeDataToDb");
@@ -122,6 +124,7 @@ export async function syncStripeDataToDb({
         pendingInvites: true,
         users: {
           select: {
+            id: true,
             email: true,
             _count: { select: { emailAccounts: true } },
           },
@@ -141,6 +144,24 @@ export async function syncStripeDataToDb({
     });
 
     await syncSeats(updatedPremium);
+
+    after(() => {
+      const userIds = updatedPremium.users.map((user) => user.id);
+
+      const statusChanged =
+        currentPremium?.stripeSubscriptionStatus !== subscription.status;
+      const tierChanged = currentPremium?.tier !== tier;
+
+      if (userIds.length && (!currentPremium || statusChanged || tierChanged)) {
+        ensureEmailAccountsWatched({ userIds }).catch((error) => {
+          logger.error("Failed to ensure email watches after Stripe sync", {
+            customerId,
+            userIds,
+            error,
+          });
+        });
+      }
+    });
   } catch (error) {
     logger.error("Error syncing Stripe data to DB", { customerId, error });
     throw error;

apps/web/utils/auth.test.ts

@@ -5,6 +5,8 @@ import { captureException } from "@/utils/error";
 import { handleReferralOnSignUp } from "@/utils/auth";
 
 // Mock the dependencies
+vi.mock("server-only", () => ({}));
+
 vi.mock("next/headers", () => ({
   cookies: vi.fn(),
 }));