feat(watcher): gate receipt watch with WATCHER_MAILBOXES

Limit Pub/Sub INBOX watch and receipt processing to an allowlisted mailbox set; leave the agent CLI usable for all connected accounts.

Co-authored-by: Cursor <cursoragent@cursor.com>

apps/watcher/src/config.js

@@ -20,6 +20,36 @@ export const GOOGLE_CLOUD_PROJECT =
 export const PUBSUB_VERIFICATION_TOKEN =
   process.env.PUBSUB_VERIFICATION_TOKEN || "";
 export const SQLITE_PATH = process.env.SQLITE_PATH || defaultSqlitePath;
+/** Optional default mailbox email when scripts omit --user */
+export const GMAIL_DEFAULT_USER = (process.env.GMAIL_DEFAULT_USER || "").trim();
+
+/**
+ * Parse comma-separated mailbox allowlist for the receipt watcher.
+ * Empty / unset → empty Set (caller treats as "allow all").
+ */
+export function parseWatcherMailboxes(raw) {
+  const set = new Set();
+  for (const part of String(raw || "").split(",")) {
+    const email = part.trim().toLowerCase();
+    if (email) set.add(email);
+  }
+  return set;
+}
+
+/** Lowercased emails allowed for Pub/Sub receipt watch. Empty = all connected accounts. */
+export const WATCHER_MAILBOXES = parseWatcherMailboxes(
+  process.env.WATCHER_MAILBOXES,
+);
+
+/**
+ * @param {string|null|undefined} email
+ * @param {Set<string>} [allowlist=WATCHER_MAILBOXES]
+ */
+export function isWatcherMailboxAllowed(email, allowlist = WATCHER_MAILBOXES) {
+  if (!allowlist || allowlist.size === 0) return true;
+  if (!email) return false;
+  return allowlist.has(String(email).trim().toLowerCase());
+}
 
 export const POCKETBASE_URL = (process.env.POCKETBASE_URL || "").replace(
   /\/$/,

apps/watcher/src/index.js

@@ -1,5 +1,12 @@
 import Fastify from "fastify";
-import { GOOGLE_PUBSUB_TOPIC, PORT, WATCH_RENEW_MS, assertConfig } from "./config.js";
+import {
+  GOOGLE_PUBSUB_TOPIC,
+  PORT,
+  WATCH_RENEW_MS,
+  WATCHER_MAILBOXES,
+  assertConfig,
+  isWatcherMailboxAllowed,
+} from "./config.js";
 import { listUsersWithTokens } from "./db.js";
 import { startWatch } from "./google.js";
 import { fetchAndLogMessage } from "./handlers/fetch-and-log-message.js";
@@ -44,7 +51,22 @@ async function renewAllWatches() {
     app.log.warn("gmail-watch: no signed-in users with tokens");
     return;
   }
+
+  if (WATCHER_MAILBOXES.size > 0) {
+    app.log.info(
+      { allowlist: [...WATCHER_MAILBOXES] },
+      "gmail-watch: WATCHER_MAILBOXES restrict receipt watches",
+    );
+  }
+
   for (const user of users) {
+    if (!isWatcherMailboxAllowed(user.email)) {
+      app.log.info(
+        { userId: user.id, email: user.email },
+        "gmail-watch: skip watch (not in WATCHER_MAILBOXES)",
+      );
+      continue;
+    }
     try {
       await startWatch(user.id);
       app.log.info(

apps/watcher/src/routes/pubsub.js

@@ -1,5 +1,5 @@
 import { timingSafeEqual } from "node:crypto";
-import { PUBSUB_VERIFICATION_TOKEN } from "../config.js";
+import { PUBSUB_VERIFICATION_TOKEN, isWatcherMailboxAllowed } from "../config.js";
 import { getUserByEmail, getWatchState, saveWatchState } from "../db.js";
 import {
   gmailClientForUser,
@@ -71,6 +71,14 @@ function decodeNotification(body) {
 }
 
 async function processNotification(log, { fetchHandlers, processHandlers }, notification) {
+  if (!isWatcherMailboxAllowed(notification.emailAddress)) {
+    log.warn(
+      { email: notification.emailAddress },
+      "gmail-watch: mailbox not in WATCHER_MAILBOXES, acking",
+    );
+    return;
+  }
+
   const user = getUserByEmail(notification.emailAddress);
   if (!user) {
     log.warn(