获取Gmail API的historyId后，如何实现Express服务器的发件通知？

我有一个Express应用，在成功获取用户认证的Google授权码后会触发登录端点。验证该授权码并调用Gmail API的watch函数后，已成功获取historyId（登录并接受Google账户权限后，从Gmail API的Pub/Sub接收的History Id），但不清楚该如何使用它。

我是不是快要实现「当已认证的Gmail账户发送邮件时，服务器能收到通知」的目标了？后续步骤是什么？我目前读不懂Google文档，核心需求是实现用户发送邮件时服务器能收到相关数据，不确定当前缺少哪些环节。

附上我的Express应用代码：

require("dotenv").config();
const { PORT } = process.env;
const { google } = require("googleapis");
const { OAuth2Client } = require("google-auth-library");

const express = require("express");

const app = express();

app.use(express.json());

const oAuth2Client = new OAuth2Client(
  process.env.GOOGLE_CLIENT_ID,
  process.env.GOOGLE_CLIENT_SECRET,
  process.env.GOOGLE_REDIRECT
);

const gmail = google.gmail({
  version: "v1",
  auth: oAuth2Client,
});

app.post(URL, async (req, res) => {
  console.log(req.body.code);

  // Now that we have the code, use that to acquire tokens.
  const r = await oAuth2Client.getToken(req.body.code);

  oAuth2Client.setCredentials(r.tokens);

  console.log("r", await r);

  const reser = await gmail.users.watch({
    userId: "me",
    requestBody: {
      topicName: `projects/express-server-connection/topics/MyTopic`,
      labelIds: ["INBOX"],
    },
  });
  console.log(reser.data);

  res.send(r);
});

app.listen(PORT, async () => {
  // setting up our webhook url on server spinup
  try {
    console.log(`Server is up and Running at PORT : ${PORT}`);
  } catch (error) {
    console.log(error.message);
  }
});

你已经完成了关键的第一步（通过watch订阅Gmail变更），离目标只差几个核心环节，以下是具体后续步骤：

1. 配置Pub/Sub推送订阅 #

你已经指定了Pub/Sub主题projects/express-server-connection/topics/MyTopic，接下来需要给该主题配置推送订阅，将Gmail的变更通知转发到你的Express服务器：

• 登录Google Cloud控制台，找到对应Pub/Sub主题
• 创建新订阅，选择「推送」类型，填写服务器公网可访问的URL（比如https://your-domain.com/gmail-webhook）
• 确保该URL能处理POST请求，且返回200状态码确认接收

2. 添加Express Webhook处理端点 #

在你的代码中新增POST端点，用于接收Pub/Sub推送的Gmail变更通知：

app.post('/gmail-webhook', async (req, res) => {
  // 验证Pub/Sub消息合法性，防止恶意请求
  const message = req.body.message;
  if (!message) {
    return res.status(400).send('无效的Pub/Sub消息');
  }

  // 解析base64编码的Gmail通知数据
  const payload = JSON.parse(Buffer.from(message.data, 'base64').toString());
  
  console.log('Gmail变更通知:', payload);

  // 必须返回200，否则Pub/Sub会重复推送消息
  res.status(200).send('OK');
});

3. 用historyId获取具体邮件数据 #

收到Pub/Sub通知后，会拿到新的historyId，结合之前保存的初始historyId，调用Gmail API的users.history.list接口，获取两段ID之间的变更记录，筛选出「发送邮件」事件：

// 在webhook处理函数中添加以下逻辑
// 注意：初始historyId要关联用户存在数据库中，不能全局共享
const initialHistoryId = 你保存的用户初始historyId;
const newHistoryId = payload.historyId;

const historyResponse = await gmail.users.history.list({
  userId: 'me',
  startHistoryId: initialHistoryId,
  historyTypes: ['messageSent'] // 仅筛选发送邮件的事件
});

// 遍历变更记录，获取邮件详情
historyResponse.data.history?.forEach(history => {
  if (history.messagesAdded) {
    history.messagesAdded.forEach(msgAdded => {
      gmail.users.messages.get({
        userId: 'me',
        id: msgAdded.message.id,
        format: 'full' // 按需选择格式：raw、full、metadata等
      }).then(msg => {
        console.log('用户发送的邮件:', msg.data);
        // 此处可处理邮件数据，如存库、触发业务逻辑等
      });
    });
  }
});

4. 关键注意事项 #

• 用户数据隔离：每个用户的OAuth tokens和初始historyId要存在数据库中，不能使用全局oAuth2Client，否则多用户会冲突
• 权限配置：确保OAuth请求包含https://www.googleapis.com/auth/gmail.modify或更高权限，否则无法调用history接口
• Pub/Sub权限：给Gmail服务账户gmail-api-push@system.gserviceaccount.com授予Pub/Sub主题的发布权限，否则Gmail无法推送消息
• 消息去重：Pub/Sub可能重复推送消息，需根据message.messageId做去重处理

内容的提问来源于stack exchange，提问作者Anders Kitson