gh-christianlouis-inboxconverge/backend/app/services/gmail_service.py

"""
Gmail API service for injecting emails directly into Gmail.

Uses the Gmail API's users.messages.insert() method to inject emails
into a user's Gmail account, preserving original headers and metadata.
This is preferred over SMTP forwarding as it doesn't modify the email.
"""

import asyncio
import base64
import logging
import textwrap
import time
from datetime import datetime, timedelta, timezone
from email.mime.text import MIMEText
from email.utils import format_datetime
from typing import Optional, Dict, Any

import google.auth.exceptions
from google.auth.transport.requests import Request as GoogleAuthRequest
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

from app.core.metrics import (
    GMAIL_API_REQUESTS_TOTAL,
    GMAIL_API_DURATION_SECONDS,
    GMAIL_TOKEN_REFRESHES_TOTAL,
)
from app.utils.gmail_labels import render_import_labels

logger = logging.getLogger(__name__)

# Gmail API scopes needed for email injection
GMAIL_SCOPES = [
    "https://www.googleapis.com/auth/gmail.insert",
    "https://www.googleapis.com/auth/gmail.labels",
    "https://www.googleapis.com/auth/gmail.readonly",
]


class GmailInjectionError(Exception):
    """Raised when Gmail API injection fails"""

    pass


class GmailAuthError(GmailInjectionError):
    """Raised when Gmail OAuth2 authentication fails (token expired or revoked)."""

    pass


class GmailService:
    """
    Service for injecting emails into Gmail via the Gmail API.

    Uses users.messages.insert() which places emails directly into
    the user's mailbox without sending them through SMTP.
    """

    def __init__(
        self,
        access_token: str,
        refresh_token: Optional[str] = None,
        token_uri: str = "https://oauth2.googleapis.com/token",
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
    ):
        """
        Initialize Gmail service with OAuth2 credentials.

        Args:
            access_token: Valid OAuth2 access token
            refresh_token: OAuth2 refresh token for automatic renewal
            token_uri: OAuth2 token endpoint
            client_id: Google OAuth2 client ID
            client_secret: Google OAuth2 client secret
        """
        self._initial_access_token = access_token
        self.credentials = Credentials(
            token=access_token,
            refresh_token=refresh_token,
            token_uri=token_uri,
            client_id=client_id,
            client_secret=client_secret,
            scopes=GMAIL_SCOPES,
        )
        self._service = None
        logger.debug(
            "OAuth [GmailService]: initialized — has_refresh_token=%s, "
            "has_client_id=%s, has_client_secret=%s",
            bool(refresh_token),
            bool(client_id),
            bool(client_secret),
        )

    @property
    def service(self):
        """Lazy-initialize the Gmail API service."""
        if self._service is None:
            self._service = build(
                "gmail", "v1", credentials=self.credentials, cache_discovery=False
            )
        return self._service

    async def inject_email(
        self,
        raw_email: bytes,
        label_ids: Optional[list] = None,
        source_account_name: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Inject a raw email into the user's Gmail account.

        Uses users.messages.insert() to place the email directly
        into the mailbox. The email appears as if it was received
        normally, preserving all original headers.

        Args:
            raw_email: Raw email bytes (RFC 2822 format)
            label_ids: Gmail label IDs to apply (defaults to ["INBOX"])
            source_account_name: Optional name for logging

        Returns:
            Dict with message id and thread id

        Raises:
            GmailInjectionError: If injection fails
        """
        if label_ids is None:
            label_ids = ["INBOX"]

        # Base64url encode the raw email
        encoded_message = base64.urlsafe_b64encode(raw_email).decode("utf-8")

        message_body = {
            "raw": encoded_message,
            "labelIds": label_ids,
        }

        loop = asyncio.get_event_loop()

        _start = time.perf_counter()
        try:
            result = await loop.run_in_executor(
                None,
                lambda: self.service.users()
                .messages()
                .insert(userId="me", body=message_body)
                .execute(),
            )

            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="inject", status="success").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="inject").observe(_dur)

            logger.info(
                f"Injected email into Gmail: id={result.get('id')}"
                f"{f' from {source_account_name}' if source_account_name else ''}"
            )

            return {
                "message_id": result.get("id"),
                "thread_id": result.get("threadId"),
                "label_ids": result.get("labelIds", []),
            }

        except HttpError as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="inject", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="inject").observe(_dur)
            error_msg = (
                f"Gmail API error: {e.reason if hasattr(e, 'reason') else str(e)}"
            )
            logger.error(error_msg)
            # Surface 401 so callers can mark credentials as invalid
            raise GmailInjectionError(error_msg)
        except google.auth.exceptions.RefreshError as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="inject", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="inject").observe(_dur)
            error_msg = (
                f"Gmail token refresh failed — the refresh token may have been revoked. "
                f"The user must re-authorise. Detail: {e}"
            )
            logger.error(error_msg)
            raise GmailAuthError(error_msg)
        except Exception as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="inject", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="inject").observe(_dur)
            error_msg = f"Failed to inject email into Gmail: {str(e)}"
            logger.error(error_msg)
            raise GmailInjectionError(error_msg)

    async def verify_access(self) -> bool:
        """
        Verify that the Gmail API credentials are valid.

        Returns:
            True if credentials are valid and can access Gmail
        """
        loop = asyncio.get_event_loop()

        _start = time.perf_counter()
        try:
            result = await loop.run_in_executor(
                None,
                lambda: self.service.users().getProfile(userId="me").execute(),
            )
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="verify", status="success").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="verify").observe(_dur)
            email = result.get("emailAddress", "unknown")
            logger.info(f"Gmail API access verified for: {email}")
            return True
        except google.auth.exceptions.RefreshError as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="verify", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="verify").observe(_dur)
            logger.error(
                f"Gmail token refresh failed during access verification — "
                f"refresh token may be revoked. Detail: {e}"
            )
            return False
        except Exception as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="verify", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="verify").observe(_dur)
            logger.error(f"Gmail API access verification failed: {e}")
            return False

    async def get_email_address(self) -> Optional[str]:
        """
        Get the email address associated with the Gmail credentials.

        Returns:
            Email address string or None if unavailable
        """
        loop = asyncio.get_event_loop()

        _start = time.perf_counter()
        try:
            result = await loop.run_in_executor(
                None,
                lambda: self.service.users().getProfile(userId="me").execute(),
            )
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(
                operation="get_profile", status="success"
            ).inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="get_profile").observe(_dur)
            email = result.get("emailAddress")
            logger.debug("OAuth [GmailService]: fetched email address=%s", email)
            return email
        except google.auth.exceptions.RefreshError as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(
                operation="get_profile", status="error"
            ).inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="get_profile").observe(_dur)
            logger.error(
                "OAuth [GmailService]: token refresh failed while fetching email "
                "address — refresh token may be revoked. Detail: %s",
                e,
            )
            return None
        except Exception as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(
                operation="get_profile", status="error"
            ).inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="get_profile").observe(_dur)
            logger.error(f"Failed to get Gmail email address: {e}")
            return None

    async def get_or_create_label(self, name: str) -> str:
        """
        Return the Gmail label ID for a label with the given name.

        Lists the user's existing labels and returns the ID of the first
        match (case-insensitive).  If no matching label is found, a new
        label is created and its ID is returned.

        Args:
            name: Human-readable label name (e.g. "test", "imported").

        Returns:
            Gmail label ID string (e.g. "Label_1234567890").

        Raises:
            GmailInjectionError: If the Gmail API call fails.
        """
        loop = asyncio.get_event_loop()

        _start = time.perf_counter()
        try:
            labels_resp = await loop.run_in_executor(
                None,
                lambda: self.service.users().labels().list(userId="me").execute(),
            )
            for label in labels_resp.get("labels", []):
                if label.get("name", "").lower() == name.lower():
                    _dur = time.perf_counter() - _start
                    GMAIL_API_REQUESTS_TOTAL.labels(
                        operation="get_label", status="success"
                    ).inc()
                    GMAIL_API_DURATION_SECONDS.labels(operation="get_label").observe(
                        _dur
                    )
                    return label["id"]

            # Label not found – create it
            created = await loop.run_in_executor(
                None,
                lambda: self.service.users()
                .labels()
                .create(userId="me", body={"name": name})
                .execute(),
            )
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(
                operation="get_label", status="success"
            ).inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="get_label").observe(_dur)
            logger.info(f"Created Gmail label '{name}' with id={created['id']}")
            return created["id"]

        except HttpError as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="get_label", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="get_label").observe(_dur)
            error_msg = f"Gmail API error while managing label '{name}': {e.reason if hasattr(e, 'reason') else str(e)}"
            logger.error(error_msg)
            raise GmailInjectionError(error_msg)
        except google.auth.exceptions.RefreshError as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="get_label", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="get_label").observe(_dur)
            error_msg = (
                f"Gmail token refresh failed while managing label '{name}' — "
                f"refresh token may be revoked. Detail: {e}"
            )
            logger.error(error_msg)
            raise GmailAuthError(error_msg)
        except Exception as e:
            _dur = time.perf_counter() - _start
            GMAIL_API_REQUESTS_TOTAL.labels(operation="get_label", status="error").inc()
            GMAIL_API_DURATION_SECONDS.labels(operation="get_label").observe(_dur)
            error_msg = f"Failed to get/create Gmail label '{name}': {str(e)}"
            logger.error(error_msg)
            raise GmailInjectionError(error_msg)

    async def inject_debug_email(
        self,
        recipient_email: str,
        import_label_templates: Optional[list[str]] = None,
    ) -> Dict[str, Any]:
        """
        Inject a debug/test email into the user's Gmail inbox.

        The message is made to appear as if it was sent by
        christian@docuelevate.org on the current date.  It is placed in
        the inbox and tagged with the custom labels "test" and "imported"
        so it is easy to identify and clean up.

        Args:
            recipient_email: The Gmail address to deliver the message to
                (the authenticated user's address).

        Returns:
            Dict with message_id, thread_id, and label_ids.

        Raises:
            GmailInjectionError: If injection or label management fails.
        """
        now = datetime.now(timezone.utc)
        date_str = now.strftime("%d %B %Y")  # e.g. "25 March 2026"

        subject = f"Test Import – {date_str}"

        body = textwrap.dedent(f"""\
            Hi there,

            This is an automated test message injected via the Gmail API to
            confirm that the import pipeline is working correctly.

            Date:   {date_str}
            Source: DocuElevate Integration Test

            If you can see this message in your inbox it means that Gmail API
            delivery is functioning as expected.  Feel free to delete it.

            Best regards,
            Christian Krakau-Louis
            DocuElevate
            """)

        msg = MIMEText(body, "plain", "utf-8")
        msg["From"] = "Christian Krakau-Louis <christian@docuelevate.org>"
        msg["To"] = recipient_email
        msg["Subject"] = subject
        msg["Date"] = format_datetime(now)
        msg["Message-ID"] = f"<debug-{now.strftime('%Y%m%d%H%M%S')}@docuelevate.org>"

        raw_bytes = msg.as_bytes()

        label_ids = await self.build_import_label_ids(import_label_templates)
        test_label_id = await self.get_or_create_label("test")
        if test_label_id not in label_ids:
            label_ids.append(test_label_id)

        return await self.inject_email(
            raw_email=raw_bytes,
            label_ids=label_ids,
            source_account_name="debug",
        )

    async def build_import_label_ids(
        self,
        import_label_templates: Optional[list[str]] = None,
        source_email: Optional[str] = None,
    ) -> list[str]:
        """Resolve configured import labels into Gmail label IDs."""
        label_ids = ["INBOX"]

        for label_name in render_import_labels(import_label_templates, source_email):
            if label_name.upper() == "INBOX":
                continue
            label_id = await self.get_or_create_label(label_name)
            if label_id not in label_ids:
                label_ids.append(label_id)

        return label_ids

    @staticmethod
    def _tz_aware_expiry(expiry: Optional[datetime]) -> Optional[datetime]:
        """Return *expiry* with UTC tzinfo attached if it is naive.

        google-auth sets ``credentials.expiry`` as a naive UTC datetime
        (``datetime.utcnow() + timedelta(...)``).  Storing a naive datetime
        into a ``DateTime(timezone=True)`` column causes silent tz-mismatch
        bugs in comparisons and storage, so we always normalise before
        returning expiry values to callers.
        """
        if expiry is not None and expiry.tzinfo is None:
            return expiry.replace(tzinfo=timezone.utc)
        return expiry

    def is_token_expiring_soon(self, within_minutes: int = 30) -> bool:
        """
        Return True if the access token has already expired or will expire
        within *within_minutes* minutes.

        When ``credentials.expiry`` is None the expiry is unknown, which is
        treated as expiring soon so a proactive refresh is performed.

        Args:
            within_minutes: Threshold in minutes before which a token is
                considered "expiring soon".

        Returns:
            True if the token needs refreshing, False otherwise.
        """
        if self.credentials.expiry is None:
            return True
        threshold = datetime.now(timezone.utc) + timedelta(minutes=within_minutes)
        expiry = self.credentials.expiry
        # google-auth stores expiry as a naive UTC datetime; make it tz-aware.
        if expiry.tzinfo is None:
            expiry = expiry.replace(tzinfo=timezone.utc)
        return expiry <= threshold

    async def proactive_refresh(self) -> Dict[str, Any]:
        """
        Explicitly refresh the access token using the stored refresh token.

        Unlike the lazy refresh that happens automatically during API calls,
        this method triggers a refresh regardless of whether the current
        access token has expired.  Use this from a scheduled task to keep
        tokens fresh and to detect revocation early.

        Returns:
            Dict with ``access_token`` and ``expiry`` (datetime | None)
            representing the newly obtained access token.

        Raises:
            GmailAuthError: If the refresh token is missing, has been revoked,
                or the refresh request fails for an auth-related reason.
            GmailInjectionError: For unexpected non-auth errors.
        """
        if not self.credentials.refresh_token:
            raise GmailAuthError(
                "Cannot refresh: no refresh token stored. "
                "Re-authorise Gmail to obtain a new refresh token."
            )

        loop = asyncio.get_event_loop()
        try:
            await loop.run_in_executor(
                None,
                lambda: self.credentials.refresh(GoogleAuthRequest()),
            )
            GMAIL_TOKEN_REFRESHES_TOTAL.inc()
            logger.info(
                "Gmail access token refreshed proactively; new expiry: %s",
                self.credentials.expiry,
            )
            return {
                "access_token": self.credentials.token,
                "expiry": self._tz_aware_expiry(self.credentials.expiry),
            }
        except google.auth.exceptions.RefreshError as e:
            error_msg = f"Gmail refresh token has been revoked or is invalid — the user must re-authorise. Detail: {e}"
            logger.error(error_msg)
            raise GmailAuthError(error_msg)
        except Exception as e:
            raise GmailInjectionError(
                f"Unexpected error during Gmail token refresh: {e}"
            )

    def get_refreshed_token(self) -> Optional[Dict[str, Any]]:
        """
        Return the current access token and expiry if the token was refreshed
        since this service instance was created.

        The google-auth library auto-refreshes the access token when an API
        call is made with an expired token.  Call this after inject_email() to
        check whether a refresh happened and persist the new token.

        Returns:
            Dict with ``access_token`` and ``expiry`` (datetime | None), or
            None if the token has not changed from the one passed to __init__.
        """
        current_token = self.credentials.token
        if current_token and current_token != self._initial_access_token:
            GMAIL_TOKEN_REFRESHES_TOTAL.inc()
            logger.debug(
                "OAuth [GmailService]: access token was auto-refreshed during API "
                "call; new expiry=%s",
                self.credentials.expiry,
            )
            return {
                "access_token": current_token,
                "expiry": self._tz_aware_expiry(self.credentials.expiry),
            }
        return None