Skip to main content

ADR-004: Push Notification Architecture for Inbox Monitoring

Status: Accepted Date: December 17, 2025 Decision Makers: CODITECT Engineering Team

Context​

CODITECT needs to monitor Gmail inboxes for new messages in real-time to enable:

  1. Meeting Follow-up Detection: Find emails related to meetings
  2. RSVP Processing: Detect meeting responses
  3. AI Email Analysis: Process incoming emails with AI
  4. Workflow Triggers: Start workflows based on emails

Gmail API offers two approaches for inbox monitoring:

  • Polling: Periodically check for new messages
  • Push Notifications: Gmail notifies us via Cloud Pub/Sub

Decision​

We will use Gmail Push Notifications via Cloud Pub/Sub as the primary mechanism for inbox monitoring, with polling as a fallback.

Architecture​

┌─────────────┐       ┌─────────────────┐       ┌─────────────────┐
│   Gmail     │──────▶│  Cloud Pub/Sub  │──────▶│  CODITECT       │
│   Inbox     │ push  │  (Topic)        │  push │  Inbox Watcher  │
└─────────────┘       └─────────────────┘       └─────────────────┘
                                                        │
                                                        ▼
                                               ┌─────────────────┐
                                               │  History API    │
                                               │  (Get Messages) │
                                               └─────────────────┘

Implementation​

Step 1: Set Up Watch​

async def start_watch(self, label_ids: List[str] = None) -> str:
    """Start Gmail push notifications."""
    result = await self.gmail_client.watch(
        topic_name=f'projects/{self.project_id}/topics/gmail-push',
        label_ids=label_ids or ['INBOX'],
    )
    self._history_id = result['historyId']
    return result['expiration']

Step 2: Receive Pub/Sub Notifications​

def _handle_pubsub_message(self, message):
    """Handle push notification from Gmail."""
    data = json.loads(message.data.decode())
    history_id = data['historyId']
    email_address = data['emailAddress']

    # Fetch new messages since last history ID
    asyncio.create_task(self._process_history(history_id))
    message.ack()

Step 3: Fetch New Messages​

async def _process_history(self, new_history_id: str):
    """Get new messages via History API."""
    history = await self.gmail_client.list_history(
        start_history_id=self._history_id,
        history_types=['messageAdded'],
    )

    for record in history.get('history', []):
        for added in record.get('messagesAdded', []):
            message = await self.gmail_client.get_message(
                added['message']['id']
            )
            await self._handle_new_message(message)

    self._history_id = new_history_id

Watch Renewal​

Gmail watches expire after 7 days. We must renew before expiration.

class WatchRenewalScheduler:
    """Automatically renew Gmail watch before expiration."""

    RENEWAL_BUFFER = timedelta(hours=24)  # Renew 1 day before expiry

    async def schedule_renewal(self, expiration: str):
        """Schedule watch renewal."""
        expiry = datetime.fromtimestamp(int(expiration) / 1000)
        renewal_time = expiry - self.RENEWAL_BUFFER

        await self.scheduler.schedule(
            task_id='gmail_watch_renewal',
            trigger_time=renewal_time,
            callback=self._renew_watch,
        )

    async def _renew_watch(self):
        """Renew the Gmail watch."""
        await self.inbox_watcher.stop()
        await self.inbox_watcher.start()

Fallback: Polling​

If Pub/Sub fails or isn't available, fall back to polling.

class PollingFallback:
    """Polling fallback for environments without Pub/Sub."""

    POLL_INTERVAL = 60  # seconds

    async def start_polling(self, label_ids: List[str] = None):
        """Start polling for new messages."""
        while self._running:
            messages = await self.gmail_client.list_messages(
                query='is:unread',
                label_ids=label_ids or ['INBOX'],
            )

            for msg in messages.get('messages', []):
                if msg['id'] not in self._seen_ids:
                    await self._handle_new_message(msg['id'])
                    self._seen_ids.add(msg['id'])

            await asyncio.sleep(self.POLL_INTERVAL)

Consequences​

Positive​

  • Real-time: Messages processed within seconds
  • Efficient: No wasted API calls polling empty inbox
  • Scalable: Pub/Sub handles high message volumes
  • Reliable: Pub/Sub guarantees at-least-once delivery

Negative​

  • Complexity: Requires Pub/Sub setup
  • GCP Dependency: Needs Google Cloud project
  • Watch Expiration: Must handle 7-day renewal
  • History Gaps: May miss messages if history ID expires

Mitigation​

  • Automated watch renewal scheduler
  • Periodic full sync to catch missed messages
  • Polling fallback for non-GCP environments
  • Clear setup documentation

Pub/Sub Setup Requirements​

# Create topic (one-time)
gcloud pubsub topics create gmail-push

# Create subscription
gcloud pubsub subscriptions create gmail-push-sub \
    --topic=gmail-push \
    --push-endpoint=https://your-domain.com/webhook/gmail

# Grant Gmail permission to publish
gcloud pubsub topics add-iam-policy-binding gmail-push \
    --member="serviceAccount:gmail-api-push@system.gserviceaccount.com" \
    --role="roles/pubsub.publisher"

Cost Analysis​

ResourceFree TierOur Usage
Pub/Sub Messages10GB/month< 1GB/month
Pub/Sub DeliveryFirst 10M deliveries< 100K/month
Total Cost$0$0
  • ADR-001: Email Provider Abstraction
  • ADR-002: Zero-Cost Architecture
  • ADR-003: OAuth Strategy

Document Control:

  • Created: December 17, 2025
  • Author: CODITECT Engineering Team