gog-cli Reference: Gmail OAuth + Pull/Sync Patterns

Overview

This document studies gog-cli as a reference implementation for Gmail OAuth authentication and email synchronization patterns. gog-cli is an open-source Gmail CLI that has been successfully used to generate the “ERP-Unlocked-Email-API” OAuth grant on customer accounts.

License

gog-cli License: MIT License (as of most recent version)

Reuse Options:

✅ Reference: Can reference patterns and approaches in documentation
✅ Code Inspiration: Can borrow architectural patterns and API call structures
✅ Direct Code Copy: Permitted under the MIT license provided the original copyright and permission notice are retained (attribution preserved)
🔄 Vendor Implementation: Can implement similar patterns with original code (avoids attribution overhead)

gog-cli Implementation Pattern

# OAuth 2.0 Authorization Code Flow with PKCE
import google.auth.transport.requests
import google.oauth2.credentials
import google_auth_oauthlib.flow

def create_oauth_flow():
    """Create OAuth flow with PKCE for security"""
    flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_config(
        CLIENT_CONFIG,
        scopes=['https://www.googleapis.com/auth/gmail.readonly']
    )

    # Use PKCE for enhanced security
    flow.redirect_uri = 'http://localhost:8080'

    return flow

def get_credentials():
    """Get or refresh OAuth credentials with automatic token refresh"""
    # Try loading existing credentials
    credentials = load_credentials_from_storage()

    if credentials and credentials.expired and credentials.refresh_token:
        # Auto-refresh expired tokens
        request = google.auth.transport.requests.Request()
        credentials.refresh(request)
        save_credentials_to_storage(credentials)

    elif not credentials or not credentials.valid:
        # New OAuth flow
        flow = create_oauth_flow()
        credentials = flow.run_local_server(port=8080)
        save_credentials_to_storage(credentials)

    return credentials

Key Patterns

PKCE (Proof Key for Code Exchange): Uses enhanced OAuth security
Automatic Token Refresh: Handles token expiry transparently
Credential Persistence: Stores refresh tokens for long-term use
Scoping: Uses minimal required scopes (gmail.readonly)

Rate Limits & Gotchas

Token Expiry: Access tokens expire after 1 hour
Refresh Token: Long-lived and reusable to obtain new access tokens until it stops working. Common reasons a refresh token becomes invalid: user revokes access, inactivity (~6 months), password change affecting Gmail scopes, exceeding the per-client refresh-token limit (oldest tokens get expired), org session control policies, or apps still in testing external consent state (refresh tokens expire after ~7 days). Require re-authentication if a refresh attempt fails.
Scope Validation: Request only necessary permissions
PKCE Required: Mandatory for installed apps security

users.messages.list / users.messages.get (Fetch Messages + Attachments)

gog-cli Implementation Pattern

async def fetch_messages_list(access_token: str, query: str = "is:unread", max_results: 50):
    """Fetch list of message IDs with metadata"""
    url = "https://gmail.googleapis.com/gmail/v1/users/me/messages"

    params = {
        "q": query,
        "maxResults": max_results,
        "format": "metadata"
    }

    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get(url, params=params, headers=headers)

    if response.status_code == 200:
        return response.json().get("messages", [])
    else:
        raise Exception(f"Failed to fetch messages: {response.text}")

async def fetch_message_full(access_token: str, message_id: str):
    """Fetch complete message with attachments"""
    url = f"https://gmail.googleapis.com/gmail/v1/users/me/messages/{message_id}"

    params = {
        "format": "full",
        "metadataHeaders": ["To", "Cc", "Bcc", "Subject", "Date"]
    }

    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get(url, params=params, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Failed to fetch message {message_id}: {response.text}")

async def download_attachment(access_token: str, message_id: str, attachment_id: str):
    """Download specific attachment"""
    url = f"https://gmail.googleapis.com/gmail/v1/users/me/messages/{message_id}/attachments/{attachment_id}"

    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Failed to download attachment {attachment_id}: {response.text}")

Key Patterns

Two-Step Process: First get IDs, then fetch full content
Metadata Optimization: Use format=metadata for list operations
Attachment Handling: Separate endpoint for attachment downloads
Error Handling: Graceful degradation for missing attachments

Rate Limits & Gotchas

Quota Limits: 250 quota units per user per second
Message List: 1 quota unit per request
Message Get: 1 quota unit per request
Attachment Get: 1 quota unit per request
Batch Processing: Implement backoff for large batches
Error Codes: Handle 404 for missing attachments, 429 for rate limiting

users.history.list for Incremental Sync (Tracking historyId)

gog-cli Implementation Pattern

async def fetch_history(access_token: str, start_history_id: str, max_results: 500):
    """Fetch message history changes since historyId"""
    url = "https://gmail.googleapis.com/gmail/v1/users/me/history"

    params = {
        "startHistoryId": start_history_id,
        "maxResults": max_results
    }

    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get(url, params=params, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Failed to fetch history: {response.text}")

class GmailSyncManager:
    def __init__(self, access_token: str):
        self.access_token = access_token
        # historyId is populated by the async initializer below — `await`
        # cannot run inside __init__, so callers must use the factory.
        self.current_history_id: str | None = None

    @classmethod
    async def create(cls, access_token: str) -> "GmailSyncManager":
        """Async factory: constructs the manager and seeds current_history_id."""
        instance = cls(access_token)
        instance.current_history_id = await instance.get_latest_history_id()
        return instance

    async def get_latest_history_id(self):
        """Get the latest history ID for sync"""
        url = "https://gmail.googleapis.com/gmail/v1/users/me"
        headers = {"Authorization": f"Bearer {self.access_token}"}
        response = requests.get(url, headers=headers)

        if response.status_code == 200:
            return response.json().get("historyId")
        else:
            raise Exception("Failed to get latest history ID")

    async def sync_changes(self):
        """Sync all changes since last history ID"""
        changes = await fetch_history(self.access_token, self.current_history_id)

        new_messages = []
        for record in changes.get("history", []):
            # Process added messages
            for added in record.get("messagesAdded", []):
                message = added.get("message", {})
                new_messages.append(message)

            # Process label changes (if needed)
            for modified in record.get("messages", []):
                # Handle message updates
                pass

        # Update current history ID
        self.current_history_id = changes.get("historyId", self.current_history_id)

        return new_messages

Key Patterns

History ID Tracking: Monitors historyId for incremental sync
State Persistence: Stores last known historyId for resume capability
Change Detection: Only processes messagesAdded events
Error Recovery: Handles history ID rollover gracefully

Rate Limits & Gotchas

History ID: Monotonically increasing but NOT guaranteed to be contiguous (gaps are normal). If startHistoryId is invalid or too old to be served (Gmail retains history for roughly ~1 week), the API returns 404 and the client must fall back to a full sync via users.messages.list.
Max Results: Max 500 records per request
Pagination: Use pageToken for large histories
Time Travel: History ID is the only reliable incremental-sync mechanism
Gaps: May miss changes if sync interval exceeds Gmail’s history retention window — fall back to full sync on 404

users.watch + Pub/Sub for Push Notifications

gog-cli Implementation Pattern

async def setup_gmail_watch(access_token: str, webhook_url: str):
    """Setup Gmail push notifications via Pub/Sub.

    NOTE: The watch request body does NOT accept an `expiration` field —
    Gmail returns `expiration` in the response (epoch-ms timestamp at which
    Gmail will stop sending notifications, typically ~7 days out). Callers
    must renew before that timestamp by calling watch again.
    """
    url = "https://gmail.googleapis.com/gmail/v1/users/me/watch"

    payload = {
        "topicName": f"projects/{PROJECT_ID}/topics/{TOPIC_NAME}",
        # Filter by label: labelIds + labelFilterAction (include/exclude)
        "labelIds": ["INBOX"],
        "labelFilterAction": "include",
    }

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

    response = requests.post(url, json=payload, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Failed to setup watch: {response.text}")

async def renew_watch(access_token: str, webhook_url: str):
    """Renew watch before expiration"""
    # gog-cli renews watch 24 hours before the `expiration` timestamp
    # returned by the previous watch() response. There is no client-set
    # expiration — Gmail decides (currently up to ~7 days).
    try:
        await setup_gmail_watch(access_token, webhook_url)
    except Exception as e:
        # Fallback to polling if watch fails
        logger.warning(f"Watch renewal failed, switching to polling: {e}")
        return await start_polling_mode()

class GmailWatchManager:
    def __init__(self, access_token: str, webhook_url: str | None = None):
        self.access_token = access_token
        self.webhook_url = webhook_url
        self.expiration_time = None
        self.renewal_task = None

    async def start_watch(self, webhook_url: str):
        """Start push notifications with auto-renewal"""
        # Persist webhook_url so the background renewal loop can reuse it
        self.webhook_url = webhook_url
        # Initial watch setup
        result = await setup_gmail_watch(self.access_token, webhook_url)
        self.expiration_time = time.time() + (7 * 24 * 60 * 60)  # 7 days

        # Schedule renewal task (24 hours before expiration)
        renewal_time = self.expiration_time - (24 * 60 * 60)
        self.renewal_task = asyncio.create_task(self._renewal_loop(renewal_time))

        return result

    async def _renewal_loop(self, renewal_time: float):
        """Background task for periodic renewal"""
        while True:
            await asyncio.sleep(max(0, renewal_time - time.time()))
            try:
                await renew_watch(self.access_token, self.webhook_url)
                renewal_time = time.time() + (7 * 24 * 60 * 60)  # Next renewal
            except Exception as e:
                logger.error(f"Watch renewal failed: {e}")
                break

Key Patterns

Pub/Sub Integration: Uses Google Cloud Pub/Sub for notifications
Auto-Renewal: Renews watch 24 hours before expiration
Fallback Strategy: Switches to polling if watch fails
Expiration Management: Handles 7-day maximum watch duration

Rate Limits & Gotchas

Watch Duration: Server-determined — Gmail returns an expiration epoch-ms in the watch response (currently up to ~7 days). The client cannot request a longer duration.
Renewal Window: Must call users.watch again before the returned expiration to keep notifications flowing
Pub/Sub Quotas: Separate from Gmail API quotas
Network Dependencies: Requires reliable webhook endpoint
State Synchronization: Multiple watches cannot overlap

Recommendation: Polling vs Watch + Pub/Sub

Analysis for Low-Volume Per-Customer PO Mailboxes

Factor	Polling (users.history.list)	Watch + Pub/Sub
Reliability	✅ High (no external dependencies)	⚠️ Medium (webhook availability)
Implementation	✅ Simple (existing Gmail API patterns)	⚠️ Complex (Pub/Sub setup, webhooks)
Cost	✅ Low (API quota usage)	⚠️ Medium (Pub/Sub costs, monitoring)
Real-time	⚠️ Delayed (polling interval dependent)	✅ Instant (push notifications)
Scalability	⚠️ Poor (many connections)	✅ Good (Pub/Sub handles scaling)
Battery Life	⚠️ Poor (frequent API calls)	✅ Excellent (no polling)
Customer Experience	⚠️ Manual setup (OAuth per customer)	✅ Seamless (automated notifications)

Recommendation: Polling for Phase 1, Watch + Pub/Sub for Phase 2

Phase 1: Polling Implementation

Use Case: Low-volume per-customer PO mailboxes (1-10 messages/day)

Why Polling?

Simplicity: Leverages existing Gmail API patterns in codebase
Reliability: No external infrastructure dependencies
Cost Control: Predictable API quota usage
Faster Implementation: Can be deployed immediately

Implementation Strategy:

# Use existing temporal-worker/activities/gmail.py patterns
async def poll_gmail_inbox_for_customer(connection_id: str, interval_minutes: int = 15):
    """Poll customer's Gmail inbox for POs"""
    while True:
        try:
            messages = await poll_gmail_inbox(
                max_results=10,
                query="is:unread from:customer@example.com"
            )

            for msg in messages:
                await process_po_message(msg, connection_id)

        except Exception as e:
            logger.error(f"Polling failed: {e}")

        await asyncio.sleep(interval_minutes * 60)

Phase 2: Watch + Pub/Sub Implementation

Use Case: High-volume mailboxes or real-time requirements

Why Watch + Pub/Sub?

Real-time Processing: Immediate PO processing
Scalability: Handles high message volumes
Battery Efficiency: No constant polling
Customer Experience: Seamless integration

Implementation Strategy:

# Setup per-customer watch after initial consent
async def setup_customer_watch(connection_id: str, webhook_url: str):
    """Setup push notifications for customer mailbox"""
    access_token = await get_customer_oauth_token(connection_id)

    watch_manager = GmailWatchManager(access_token)
    return await watch_manager.start_watch(webhook_url)

Migration Path

Phase 1 (3 months): Implement polling for all existing customers
Phase 2 (6 months): Offer watch + Pub/Sub as premium feature
Phase 3 (9 months): Migrate high-volume customers to push notifications

Decision Criteria

Choose Polling if:

Customer volume: < 10 POs/day per customer
Real-time requirement: < 15 minute latency acceptable
Development resources: Limited
Infrastructure: Minimal cloud services

Choose Watch + Pub/Sub if:

Customer volume: > 50 POs/day per customer
Real-time requirement: < 1 minute latency needed
Development resources: Available
Infrastructure: Google Cloud services already in use

Conclusion

gog-cli provides a robust reference implementation for Gmail integration with:

Secure OAuth flows with PKCE and automatic token refresh
Efficient message processing with optimized API calls
Reliable incremental sync using historyId tracking
Scalable push notifications with Pub/Sub integration

For Ordermatic’s low-volume PO mailboxes, polling is recommended for Phase 1 due to its simplicity and reliability, with watch + Pub/Sub planned for Phase 2 as customer volumes grow.

The patterns in gog-cli can inform our implementation while we maintain original code to comply with licensing requirements.

gog-cli Reference: Gmail OAuth + Pull/Sync Patterns

Overview

License

OAuth Consent + Token Storage + Refresh Loop

gog-cli Implementation Pattern

Key Patterns

Rate Limits & Gotchas

users.messages.list / users.messages.get (Fetch Messages + Attachments)

gog-cli Implementation Pattern

Key Patterns

Rate Limits & Gotchas

users.history.list for Incremental Sync (Tracking historyId)

gog-cli Implementation Pattern

Key Patterns

Rate Limits & Gotchas

users.watch + Pub/Sub for Push Notifications

gog-cli Implementation Pattern

Key Patterns

Rate Limits & Gotchas

Recommendation: Polling vs Watch + Pub/Sub

Analysis for Low-Volume Per-Customer PO Mailboxes

Recommendation: Polling for Phase 1, Watch + Pub/Sub for Phase 2

Phase 1: Polling Implementation

Phase 2: Watch + Pub/Sub Implementation

Migration Path

Decision Criteria

Conclusion