`app/api/cron/gmail-sync/route.ts`

314 lines
import { NextRequest, NextResponse } from 'next/server';
import { query } from '@/lib/db';
import { getGmailClientForMailbox, getHeader, decodeBody, decodeHtmlBody } from '@/lib/gmail';
import { verifyCronAuth } from '@/lib/cron-auth';

/**
 * GET /api/cron/gmail-sync
 * Automated Gmail sync for all active mailboxes.
 * Called by system cron every 15 minutes during business hours.
 */

// Rate limit: 200ms delay between individual message fetches to avoid Gmail API throttling
const MESSAGE_FETCH_DELAY_MS = 200;

function sleep(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms));
}

export async function GET(request: NextRequest) {
  const auth = verifyCronAuth(request);
  if (auth instanceof NextResponse) return auth;

  console.log('[gmail-sync] Cron sync starting...');

  try {
    // Get all active mailboxes that don't need reauth
    const mbRes = await query(
      `SELECT id, email, sync_cursor, last_synced_at, oauth_tokens, needs_reauth
       FROM mailboxes
       WHERE is_active = true AND (needs_reauth IS NULL OR needs_reauth = false)
       ORDER BY email`,
    );

    if (mbRes.rows.length === 0) {
      console.log('[gmail-sync] No active mailboxes found');
      return NextResponse.json({
        success: true,
        synced: 0,
        errors: [],
        mailboxes: 0,
        details: [],
      });
    }

    const details: Array<{
      email: string;
      mailboxId: string;
      fetched: number;
      inserted: number;
      skipped: number;
      error?: string;
    }> = [];
    const errors: string[] = [];
    let totalSynced = 0;

    for (const mailbox of mbRes.rows) {
      const logId = await createSyncLog(mailbox.id);

      try {
        // Check if mailbox has OAuth tokens
        if (!mailbox.oauth_tokens?.refresh_token) {
          const errMsg = `Mailbox ${mailbox.email}: No OAuth tokens / refresh token`;
          console.warn(`[gmail-sync] ${errMsg}`);
          errors.push(errMsg);

          // Mark needs reauth
          await query(
            `UPDATE mailboxes SET needs_reauth = true, updated_at = now() WHERE id = $1`,
            [mailbox.id],
          );
          await completeSyncLog(logId, 0, 'error', errMsg);
          details.push({
            email: mailbox.email,
            mailboxId: mailbox.id,
            fetched: 0,
            inserted: 0,
            skipped: 0,
            error: errMsg,
          });
          continue;
        }

        // Get Gmail client for this specific mailbox
        const gmail = await getGmailClientForMailbox(mailbox.id, mailbox.oauth_tokens);
        if (!gmail) {
          const errMsg = `Mailbox ${mailbox.email}: Failed to create Gmail client`;
          console.error(`[gmail-sync] ${errMsg}`);
          errors.push(errMsg);
          await query(
            `UPDATE mailboxes SET needs_reauth = true, updated_at = now() WHERE id = $1`,
            [mailbox.id],
          );
          await completeSyncLog(logId, 0, 'error', errMsg);
          details.push({
            email: mailbox.email,
            mailboxId: mailbox.id,
            fetched: 0,
            inserted: 0,
            skipped: 0,
            error: errMsg,
          });
          continue;
        }

        // Build Gmail query from cursor
        const cursor = mailbox.sync_cursor;
        let q = '';
        if (cursor) {
          q = `after:${cursor.replace(/-/g, '/')}`;
        }

        const maxResults = 100; // Reasonable limit per cron run
        let fetched = 0;
        let inserted = 0;
        let skipped = 0;
        let pageToken: string | undefined;
        let newestDate: string | null = null;
        let msgErrors = 0;

        // Page through messages
        do {
          const list = await gmail.users.messages.list({
            userId: 'me',
            maxResults: Math.min(maxResults - fetched, 50),
            q: q || undefined,
            pageToken,
          });

          const messageIds = list.data.messages || [];
          if (messageIds.length === 0) break;

          pageToken = list.data.nextPageToken || undefined;

          for (const stub of messageIds) {
            if (fetched >= maxResults) break;
            fetched++;

            try {
              // Rate limiting between individual message fetches
              if (fetched > 1) {
                await sleep(MESSAGE_FETCH_DELAY_MS);
              }

              const detail = await gmail.users.messages.get({
                userId: 'me',
                id: stub.id!,
                format: 'full',
              });

              const headers = detail.data.payload?.headers;
              const bodyText = decodeBody(detail.data.payload!);
              const bodyHtml = decodeHtmlBody(detail.data.payload!);
              const dateSent = getHeader(headers, 'Date');
              const toRaw = getHeader(headers, 'To');
              const ccRaw = getHeader(headers, 'Cc');

              // Track newest for cursor
              if (detail.data.internalDate && !newestDate) {
                const ms = parseInt(detail.data.internalDate, 10);
                if (!isNaN(ms)) {
                  newestDate = new Date(ms).toISOString().split('T')[0];
                }
              }

              const result = await query(
                `INSERT INTO gmail_messages
                   (mailbox_id, gmail_id, thread_id, subject, from_address, to_addresses,
                    cc_addresses, date_sent, snippet, body_text, body_html, labels, attachments, is_read)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
                 ON CONFLICT (mailbox_id, gmail_id) DO UPDATE SET
                   body_text = COALESCE(EXCLUDED.body_text, gmail_messages.body_text),
                   body_html = COALESCE(EXCLUDED.body_html, gmail_messages.body_html)
                 RETURNING id`,
                [
                  mailbox.id,
                  detail.data.id,
                  detail.data.threadId,
                  getHeader(headers, 'Subject'),
                  getHeader(headers, 'From'),
                  toRaw ? toRaw.split(',').map((a: string) => a.trim()) : [],
                  ccRaw ? ccRaw.split(',').map((a: string) => a.trim()) : [],
                  dateSent ? new Date(dateSent) : null,
                  detail.data.snippet,
                  bodyText || null,
                  bodyHtml || null,
                  detail.data.labelIds || [],
                  JSON.stringify([]),
                  !(detail.data.labelIds || []).includes('UNREAD'),
                ],
              );

              if (result.rowCount && result.rowCount > 0) {
                inserted++;
              } else {
                skipped++;
              }
            } catch (err) {
              msgErrors++;
              console.error(
                `[gmail-sync] ${mailbox.email} - Failed message ${stub.id}:`,
                (err as Error).message,
              );
            }
          }

          if (fetched >= maxResults) break;
        } while (pageToken);

        // Update cursor and last_synced_at
        if (newestDate) {
          await query(
            `UPDATE mailboxes SET last_synced_at = now(), sync_cursor = $2, updated_at = now(), needs_reauth = false WHERE id = $1`,
            [mailbox.id, newestDate],
          );
        } else {
          await query(
            `UPDATE mailboxes SET last_synced_at = now(), updated_at = now(), needs_reauth = false WHERE id = $1`,
            [mailbox.id],
          );
        }

        totalSynced += inserted;
        await completeSyncLog(logId, inserted, 'success', null);

        console.log(
          `[gmail-sync] ${mailbox.email}: fetched=${fetched} inserted=${inserted} skipped=${skipped} errors=${msgErrors}`,
        );

        details.push({
          email: mailbox.email,
          mailboxId: mailbox.id,
          fetched,
          inserted,
          skipped,
        });

        if (msgErrors > 0) {
          errors.push(`${mailbox.email}: ${msgErrors} message-level errors`);
        }
      } catch (err) {
        const errMsg = `${mailbox.email}: ${(err as Error).message}`;
        console.error(`[gmail-sync] Mailbox sync failed:`, errMsg);
        errors.push(errMsg);

        // Check for auth-related errors
        const errorMessage = (err as Error).message || '';
        if (
          errorMessage.includes('invalid_grant') ||
          errorMessage.includes('Token has been expired') ||
          errorMessage.includes('Token has been revoked') ||
          errorMessage.includes('401')
        ) {
          await query(
            `UPDATE mailboxes SET needs_reauth = true, updated_at = now() WHERE id = $1`,
            [mailbox.id],
          );
        }

        await completeSyncLog(logId, 0, 'error', (err as Error).message);
        details.push({
          email: mailbox.email,
          mailboxId: mailbox.id,
          fetched: 0,
          inserted: 0,
          skipped: 0,
          error: (err as Error).message,
        });
      }
    }

    console.log(
      `[gmail-sync] Cron complete: ${details.length} mailboxes, ${totalSynced} messages synced, ${errors.length} errors`,
    );

    return NextResponse.json({
      success: true,
      synced: totalSynced,
      errors,
      mailboxes: details.length,
      details,
    });
  } catch (err) {
    console.error('[gmail-sync] Cron fatal error:', (err as Error).message);
    return NextResponse.json(
      { error: `Gmail sync cron failed: ${(err as Error).message}` },
      { status: 500 },
    );
  }
}

/* ─── Sync Log Helpers ──────────────────────────────────────────────────── */

async function createSyncLog(mailboxId: string): Promise<string> {
  const result = await query(
    `INSERT INTO gmail_sync_log (mailbox_id, status) VALUES ($1, 'running') RETURNING id`,
    [mailboxId],
  );
  return result.rows[0].id;
}

async function completeSyncLog(
  logId: string,
  messagesSynced: number,
  status: 'success' | 'error',
  error: string | null,
): Promise<void> {
  await query(
    `UPDATE gmail_sync_log
     SET completed_at = now(), messages_synced = $2, status = $3, error = $4
     WHERE id = $1`,
    [logId, messagesSynced, status, error],
  );
}