feat: add Gmail Pub/Sub watcher for inbound email automation
New ChannelAdapter that monitors Gmail via Google Cloud Pub/Sub push notifications with polling fallback. Supports OAuth2 auth, configurable watch labels, template rendering with email metadata placeholders (from, to, subject, snippet, date, id, labels). Wired into daemon lifecycle and gateway (POST /gmail/push endpoint). Includes 16 tests covering auth, templates, push notifications, and channel routing.
This commit is contained in:
William Valentin
@@ -0,0 +1,422 @@
|
||||
import { google, type Auth } from 'googleapis';
|
||||
import { readFileSync, writeFileSync, existsSync, mkdirSync, chmodSync } from 'fs';
|
||||
import { dirname, resolve } from 'path';
|
||||
import { homedir } from 'os';
|
||||
import type { GmailConfig } from '../config/schema.js';
|
||||
import type { ChannelAdapter, ChannelStatus, InboundMessage, OutboundMessage } from '../channels/types.js';
|
||||
import { parseInterval } from './heartbeat.js';
|
||||
|
||||
/** Minimal interface for the parts of ChannelRegistry we need. */
|
||||
interface ChannelLookup {
|
||||
get(name: string): { send(peerId: string, message: OutboundMessage): Promise<void> } | undefined;
|
||||
}
|
||||
|
||||
/** Parsed email information used for template rendering. */
|
||||
interface EmailInfo {
|
||||
id: string;
|
||||
from: string;
|
||||
to: string;
|
||||
subject: string;
|
||||
snippet: string;
|
||||
date: string;
|
||||
labels: string[];
|
||||
}
|
||||
|
||||
/** Pub/Sub push notification payload from Google. */
|
||||
interface PubSubNotification {
|
||||
emailAddress: string;
|
||||
historyId: string;
|
||||
}
|
||||
|
||||
// Google Cloud Pub/Sub topic for Gmail push notifications.
|
||||
// This must be pre-configured in Google Cloud Console.
|
||||
const GMAIL_PUBSUB_TOPIC = 'projects/flynn-agent/topics/gmail-push';
|
||||
|
||||
// Watch expires after ~7 days; renew at 6 days (in ms).
|
||||
const WATCH_RENEWAL_MS = 6 * 24 * 60 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* GmailWatcher monitors a Gmail inbox for new messages and forwards them
|
||||
* as InboundMessages via the channel adapter pattern.
|
||||
*
|
||||
* Supports two modes:
|
||||
* - **Pub/Sub push**: Google sends push notifications when new emails arrive.
|
||||
* Requires a POST /gmail/push route on the gateway.
|
||||
* - **Polling fallback**: Periodically polls Gmail History API for changes.
|
||||
*
|
||||
* Authentication uses OAuth2 with a stored refresh token.
|
||||
*/
|
||||
export class GmailWatcher implements ChannelAdapter {
|
||||
readonly name = 'gmail';
|
||||
private _status: ChannelStatus = 'disconnected';
|
||||
private messageHandler?: (msg: InboundMessage) => void;
|
||||
private oauth2Client?: Auth.OAuth2Client;
|
||||
private lastHistoryId?: string;
|
||||
private pollTimer?: ReturnType<typeof setInterval>;
|
||||
private watchTimer?: ReturnType<typeof setInterval>;
|
||||
private readonly config: NonNullable<GmailConfig>;
|
||||
|
||||
constructor(
|
||||
config: NonNullable<GmailConfig>,
|
||||
private readonly channelLookup: ChannelLookup,
|
||||
) {
|
||||
this.config = config;
|
||||
}
|
||||
|
||||
get status(): ChannelStatus {
|
||||
return this._status;
|
||||
}
|
||||
|
||||
async connect(): Promise<void> {
|
||||
this._status = 'connecting';
|
||||
|
||||
try {
|
||||
this.oauth2Client = await this.authorize();
|
||||
} catch (error) {
|
||||
const errMsg = error instanceof Error ? error.message : 'Unknown error';
|
||||
console.error(`GmailWatcher: Authorization failed — ${errMsg}`);
|
||||
console.error('GmailWatcher: Run "flynn gmail-auth" to set up Gmail credentials.');
|
||||
this._status = 'error';
|
||||
return;
|
||||
}
|
||||
|
||||
// Set up Gmail push watch (Pub/Sub)
|
||||
try {
|
||||
await this.setupWatch();
|
||||
} catch (error) {
|
||||
const errMsg = error instanceof Error ? error.message : 'Unknown error';
|
||||
console.warn(`GmailWatcher: Watch setup failed (will use polling only) — ${errMsg}`);
|
||||
}
|
||||
|
||||
// Start polling fallback
|
||||
const pollMs = parseInterval(this.config.poll_interval ?? '60s');
|
||||
this.pollTimer = setInterval(() => {
|
||||
this.pollForNewMessages().catch((err) => {
|
||||
console.error('GmailWatcher: Poll error —', err instanceof Error ? err.message : err);
|
||||
});
|
||||
}, pollMs);
|
||||
|
||||
this._status = 'connected';
|
||||
console.log(`GmailWatcher: Connected (poll_interval=${this.config.poll_interval ?? '60s'})`);
|
||||
}
|
||||
|
||||
async disconnect(): Promise<void> {
|
||||
if (this.pollTimer) {
|
||||
clearInterval(this.pollTimer);
|
||||
this.pollTimer = undefined;
|
||||
}
|
||||
if (this.watchTimer) {
|
||||
clearInterval(this.watchTimer);
|
||||
this.watchTimer = undefined;
|
||||
}
|
||||
this.oauth2Client = undefined;
|
||||
this._status = 'disconnected';
|
||||
}
|
||||
|
||||
async send(peerId: string, message: OutboundMessage): Promise<void> {
|
||||
// Route responses to the configured output channel
|
||||
const outputAdapter = this.channelLookup.get(this.config.output.channel);
|
||||
if (!outputAdapter) {
|
||||
console.warn(`GmailWatcher: Output channel '${this.config.output.channel}' not found`);
|
||||
return;
|
||||
}
|
||||
|
||||
await outputAdapter.send(this.config.output.peer, message);
|
||||
}
|
||||
|
||||
onMessage(handler: (msg: InboundMessage) => void): void {
|
||||
this.messageHandler = handler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle a Pub/Sub push notification from Google.
|
||||
* Called by the gateway when POST /gmail/push is received.
|
||||
* @param data Base64-encoded Pub/Sub message data
|
||||
*/
|
||||
async handlePushNotification(data: string): Promise<void> {
|
||||
try {
|
||||
const decoded = Buffer.from(data, 'base64').toString('utf-8');
|
||||
const notification = JSON.parse(decoded) as PubSubNotification;
|
||||
|
||||
if (!notification.historyId) {
|
||||
console.warn('GmailWatcher: Push notification missing historyId');
|
||||
return;
|
||||
}
|
||||
|
||||
// Only process if the new historyId is greater than our last known one
|
||||
if (this.lastHistoryId && BigInt(notification.historyId) <= BigInt(this.lastHistoryId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.lastHistoryId) {
|
||||
await this.processHistoryChanges(this.lastHistoryId);
|
||||
}
|
||||
|
||||
this.lastHistoryId = notification.historyId;
|
||||
} catch (error) {
|
||||
console.error('GmailWatcher: Push notification error —', error instanceof Error ? error.message : error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Authorize with Gmail using stored OAuth2 credentials.
|
||||
* Reads client credentials and stored token from config paths.
|
||||
*/
|
||||
private async authorize(): Promise<Auth.OAuth2Client> {
|
||||
const credentialsPath = this.config.credentials_file;
|
||||
if (!credentialsPath) {
|
||||
throw new Error('No credentials_file configured. Set automation.gmail.credentials_file in config.');
|
||||
}
|
||||
|
||||
const expandedCredsPath = this.expandPath(credentialsPath);
|
||||
if (!existsSync(expandedCredsPath)) {
|
||||
throw new Error(`Credentials file not found: ${expandedCredsPath}`);
|
||||
}
|
||||
|
||||
const credentials = JSON.parse(readFileSync(expandedCredsPath, 'utf-8'));
|
||||
const { client_id, client_secret, redirect_uris } = credentials.installed ?? credentials.web ?? {};
|
||||
|
||||
if (!client_id || !client_secret) {
|
||||
throw new Error('Invalid credentials file — missing client_id or client_secret');
|
||||
}
|
||||
|
||||
const oauth2Client = new google.auth.OAuth2(
|
||||
client_id,
|
||||
client_secret,
|
||||
redirect_uris?.[0] ?? 'http://localhost',
|
||||
);
|
||||
|
||||
// Load stored token
|
||||
const tokenPath = this.expandPath(this.config.token_file ?? '~/.config/flynn/gmail-token.json');
|
||||
if (!existsSync(tokenPath)) {
|
||||
throw new Error(
|
||||
`Token file not found: ${tokenPath}. Run "flynn gmail-auth" to authenticate.`,
|
||||
);
|
||||
}
|
||||
|
||||
const token = JSON.parse(readFileSync(tokenPath, 'utf-8'));
|
||||
oauth2Client.setCredentials(token);
|
||||
|
||||
// Auto-save refreshed tokens
|
||||
oauth2Client.on('tokens', (newTokens) => {
|
||||
const merged = { ...token, ...newTokens };
|
||||
this.saveToken(merged);
|
||||
});
|
||||
|
||||
return oauth2Client;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up Gmail Pub/Sub watch for push notifications.
|
||||
* Calls gmail.users.watch() and schedules renewal before expiry.
|
||||
*/
|
||||
private async setupWatch(): Promise<void> {
|
||||
if (!this.oauth2Client) return;
|
||||
|
||||
const gmail = google.gmail({ version: 'v1', auth: this.oauth2Client });
|
||||
|
||||
const watchResponse = await gmail.users.watch({
|
||||
userId: 'me',
|
||||
requestBody: {
|
||||
labelIds: this.config.watch_labels ?? ['INBOX'],
|
||||
topicName: GMAIL_PUBSUB_TOPIC,
|
||||
},
|
||||
});
|
||||
|
||||
if (watchResponse.data.historyId) {
|
||||
this.lastHistoryId = watchResponse.data.historyId.toString();
|
||||
}
|
||||
|
||||
console.log(`GmailWatcher: Watch registered (historyId=${this.lastHistoryId})`);
|
||||
|
||||
// Schedule renewal before watch expiry (~7 days)
|
||||
this.watchTimer = setInterval(() => {
|
||||
this.setupWatch().catch((err) => {
|
||||
console.error('GmailWatcher: Watch renewal failed —', err instanceof Error ? err.message : err);
|
||||
});
|
||||
}, WATCH_RENEWAL_MS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Poll Gmail History API for new messages since lastHistoryId.
|
||||
* Fallback mechanism when Pub/Sub push is not available.
|
||||
*/
|
||||
private async pollForNewMessages(): Promise<void> {
|
||||
if (!this.oauth2Client) return;
|
||||
|
||||
const gmail = google.gmail({ version: 'v1', auth: this.oauth2Client });
|
||||
|
||||
// If no historyId yet, initialize it from the profile
|
||||
if (!this.lastHistoryId) {
|
||||
try {
|
||||
const profile = await gmail.users.getProfile({ userId: 'me' });
|
||||
if (profile.data.historyId) {
|
||||
this.lastHistoryId = profile.data.historyId.toString();
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('GmailWatcher: Failed to get profile —', error instanceof Error ? error.message : error);
|
||||
}
|
||||
return; // First poll — just establish the baseline
|
||||
}
|
||||
|
||||
await this.processHistoryChanges(this.lastHistoryId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch history changes since the given historyId and process new messages.
|
||||
* Updates lastHistoryId to the latest value from the response.
|
||||
*/
|
||||
private async processHistoryChanges(startHistoryId: string): Promise<void> {
|
||||
if (!this.oauth2Client) return;
|
||||
|
||||
const gmail = google.gmail({ version: 'v1', auth: this.oauth2Client });
|
||||
|
||||
try {
|
||||
const historyResponse = await gmail.users.history.list({
|
||||
userId: 'me',
|
||||
startHistoryId: startHistoryId,
|
||||
labelId: (this.config.watch_labels ?? ['INBOX'])[0],
|
||||
historyTypes: ['messageAdded'],
|
||||
});
|
||||
|
||||
const history = historyResponse.data.history ?? [];
|
||||
const processedIds = new Set<string>();
|
||||
|
||||
for (const record of history) {
|
||||
const addedMessages = record.messagesAdded ?? [];
|
||||
for (const added of addedMessages) {
|
||||
const messageId = added.message?.id;
|
||||
if (!messageId || processedIds.has(messageId)) continue;
|
||||
processedIds.add(messageId);
|
||||
|
||||
const email = await this.getMessageDetails(messageId);
|
||||
if (!email) continue;
|
||||
|
||||
// Skip messages before history_start if configured
|
||||
if (this.config.history_start) {
|
||||
const emailDate = new Date(email.date);
|
||||
const startDate = new Date(this.config.history_start);
|
||||
if (emailDate < startDate) continue;
|
||||
}
|
||||
|
||||
const text = this.renderTemplate(email);
|
||||
|
||||
const msg: InboundMessage = {
|
||||
id: `gmail-${email.id}-${Date.now()}`,
|
||||
channel: 'gmail',
|
||||
senderId: email.from,
|
||||
senderName: `gmail:${email.from}`,
|
||||
text,
|
||||
timestamp: Date.now(),
|
||||
metadata: {
|
||||
emailId: email.id,
|
||||
from: email.from,
|
||||
to: email.to,
|
||||
subject: email.subject,
|
||||
labels: email.labels,
|
||||
},
|
||||
};
|
||||
|
||||
this.messageHandler?.(msg);
|
||||
}
|
||||
}
|
||||
|
||||
// Update historyId to the latest
|
||||
if (historyResponse.data.historyId) {
|
||||
this.lastHistoryId = historyResponse.data.historyId.toString();
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
// 404 means historyId is too old — reset by fetching profile
|
||||
if (error instanceof Error && 'code' in error && (error as { code: number }).code === 404) {
|
||||
console.warn('GmailWatcher: History expired, re-syncing...');
|
||||
try {
|
||||
const profile = await gmail.users.getProfile({ userId: 'me' });
|
||||
if (profile.data.historyId) {
|
||||
this.lastHistoryId = profile.data.historyId.toString();
|
||||
}
|
||||
} catch (profileError) {
|
||||
console.error('GmailWatcher: Failed to re-sync profile —', profileError instanceof Error ? profileError.message : profileError);
|
||||
}
|
||||
} else {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch full message details by ID and extract relevant headers.
|
||||
*/
|
||||
private async getMessageDetails(messageId: string): Promise<EmailInfo | null> {
|
||||
if (!this.oauth2Client) return null;
|
||||
|
||||
const gmail = google.gmail({ version: 'v1', auth: this.oauth2Client });
|
||||
|
||||
try {
|
||||
const msg = await gmail.users.messages.get({
|
||||
userId: 'me',
|
||||
id: messageId,
|
||||
format: 'metadata',
|
||||
metadataHeaders: ['From', 'To', 'Subject', 'Date'],
|
||||
});
|
||||
|
||||
const headers = msg.data.payload?.headers ?? [];
|
||||
const getHeader = (name: string): string =>
|
||||
headers.find(h => h.name?.toLowerCase() === name.toLowerCase())?.value ?? '';
|
||||
|
||||
return {
|
||||
id: messageId,
|
||||
from: getHeader('From'),
|
||||
to: getHeader('To'),
|
||||
subject: getHeader('Subject'),
|
||||
snippet: msg.data.snippet ?? '',
|
||||
date: getHeader('Date'),
|
||||
labels: msg.data.labelIds ?? [],
|
||||
};
|
||||
} catch (error) {
|
||||
console.error(`GmailWatcher: Failed to fetch message ${messageId} —`, error instanceof Error ? error.message : error);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Render the message template with email placeholders.
|
||||
* Supported: {{from}}, {{to}}, {{subject}}, {{snippet}}, {{date}}, {{id}}, {{labels}}
|
||||
*/
|
||||
renderTemplate(email: EmailInfo): string {
|
||||
return this.config.message
|
||||
.replace(/\{\{from\}\}/g, email.from)
|
||||
.replace(/\{\{to\}\}/g, email.to)
|
||||
.replace(/\{\{subject\}\}/g, email.subject)
|
||||
.replace(/\{\{snippet\}\}/g, email.snippet)
|
||||
.replace(/\{\{date\}\}/g, email.date)
|
||||
.replace(/\{\{id\}\}/g, email.id)
|
||||
.replace(/\{\{labels\}\}/g, email.labels.join(', '));
|
||||
}
|
||||
|
||||
/**
|
||||
* Expand ~ to the user's home directory.
|
||||
*/
|
||||
expandPath(p: string): string {
|
||||
if (p.startsWith('~/') || p === '~') {
|
||||
return resolve(homedir(), p.slice(2));
|
||||
}
|
||||
return resolve(p);
|
||||
}
|
||||
|
||||
/**
|
||||
* Save token to disk with restrictive permissions (0o600).
|
||||
*/
|
||||
private saveToken(token: unknown): void {
|
||||
const tokenPath = this.expandPath(this.config.token_file ?? '~/.config/flynn/gmail-token.json');
|
||||
const dir = dirname(tokenPath);
|
||||
if (!existsSync(dir)) {
|
||||
mkdirSync(dir, { recursive: true });
|
||||
}
|
||||
writeFileSync(tokenPath, JSON.stringify(token, null, 2), 'utf-8');
|
||||
try {
|
||||
chmodSync(tokenPath, 0o600);
|
||||
} catch {
|
||||
// chmod may fail on some filesystems — not critical
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user