Skip to main content

project-document-management-docs-000-11-google-workspace-integration

title: Google Workspace Integration Specification type: reference version: 1.0.0 created: '2025-12-27' updated: '2025-12-27' status: draft tags:

  • ai-ml
  • authentication
  • security
  • testing
  • api
  • architecture
  • automation
  • backend summary: 'Google Workspace Integration Specification Document ID: 000.11 Version: 1.0.0 Status: Draft Last Updated: December 19, 2025 Owner: CODITECT Document Management Team --- Executive Summary This specification defines the integration architecture...' moe_confidence: 1.000 moe_classified: 2025-12-29

Google Workspace Integration Specification

Document ID: 000.11 Version: 1.0.0 Status: Draft Last Updated: December 19, 2025 Owner: CODITECT Document Management Team

1. Executive Summary​

This specification defines the integration architecture between the CODITECT Document Management System and Google Workspace (Drive, Docs, Sheets, Slides) and Google Cloud Platform (Cloud Storage, Cloud Identity).

1.1 Integration Scope​

ComponentIntegration LevelPriority
Google Drive APIFull CRUD + SearchP0
Shared DrivesTeam ownership modelP0
Google Docs APICreate, edit, exportP0
Google Sheets APIData binding, reportsP1
Google Slides APIPresentation generationP2
Cloud StorageBackup, archive, lifecycleP0
Cloud IdentitySSO, user provisioningP0
Admin SDKAudit logs, complianceP1

1.2 Key Benefits​

  • Seamless UX: Users work in familiar Google Workspace environment
  • Data Sovereignty: Documents stored in customer's Google environment
  • Enterprise Security: Leverages Google's security infrastructure
  • Offline Support: Google Drive desktop app provides offline access
  • Cost Efficiency: No additional storage costs for Google Workspace customers

2. Authentication & Authorization​

2.1 OAuth 2.0 Flow​

class GoogleOAuthConfig:
    """OAuth 2.0 configuration for Google Workspace integration."""

    # OAuth scopes required
    SCOPES = [
        # Drive API
        "https://www.googleapis.com/auth/drive",              # Full Drive access
        "https://www.googleapis.com/auth/drive.file",         # Per-file access
        "https://www.googleapis.com/auth/drive.metadata",     # Metadata only

        # Docs/Sheets/Slides APIs
        "https://www.googleapis.com/auth/documents",          # Google Docs
        "https://www.googleapis.com/auth/spreadsheets",       # Google Sheets
        "https://www.googleapis.com/auth/presentations",      # Google Slides

        # User info
        "https://www.googleapis.com/auth/userinfo.email",
        "https://www.googleapis.com/auth/userinfo.profile",

        # Admin (for enterprise)
        "https://www.googleapis.com/auth/admin.directory.user.readonly",
    ]

    # OAuth endpoints
    AUTH_URI = "https://accounts.google.com/o/oauth2/auth"
    TOKEN_URI = "https://oauth2.googleapis.com/token"
    REVOKE_URI = "https://oauth2.googleapis.com/revoke"

    # Token storage
    TOKEN_ENCRYPTION_KEY = os.environ["GOOGLE_TOKEN_ENCRYPTION_KEY"]

2.2 Service Account for Backend Operations​

class ServiceAccountConfig:
    """Service account for server-to-server operations."""

    # Service account credentials (stored in Secret Manager)
    CREDENTIALS_SECRET = "projects/coditect/secrets/google-sa-credentials"

    # Domain-wide delegation
    DELEGATED_ADMIN_EMAIL = "admin@customer-domain.com"

    # Scopes for service account
    SA_SCOPES = [
        "https://www.googleapis.com/auth/drive",
        "https://www.googleapis.com/auth/admin.directory.user.readonly",
        "https://www.googleapis.com/auth/admin.reports.audit.readonly",
    ]

    def get_delegated_credentials(self, user_email: str):
        """Get credentials delegated to specific user."""
        credentials = service_account.Credentials.from_service_account_info(
            self.get_credentials(),
            scopes=self.SA_SCOPES
        )
        return credentials.with_subject(user_email)

2.3 Token Management​

class GoogleTokenManager:
    """Secure token storage and refresh."""

    async def store_tokens(self, user_id: UUID, tokens: dict):
        """Store encrypted OAuth tokens."""
        encrypted = self.encrypt(json.dumps(tokens))
        await self.db.execute(
            """
            INSERT INTO google_tokens (user_id, encrypted_tokens, expires_at)
            VALUES ($1, $2, $3)
            ON CONFLICT (user_id) DO UPDATE
            SET encrypted_tokens = $2, expires_at = $3, updated_at = NOW()
            """,
            user_id, encrypted, tokens["expiry"]
        )

    async def get_valid_credentials(self, user_id: UUID) -> Credentials:
        """Get valid credentials, refreshing if needed."""
        tokens = await self.get_tokens(user_id)

        credentials = Credentials(
            token=tokens["access_token"],
            refresh_token=tokens["refresh_token"],
            token_uri=GoogleOAuthConfig.TOKEN_URI,
            client_id=os.environ["GOOGLE_CLIENT_ID"],
            client_secret=os.environ["GOOGLE_CLIENT_SECRET"]
        )

        if credentials.expired:
            credentials.refresh(Request())
            await self.store_tokens(user_id, {
                "access_token": credentials.token,
                "refresh_token": credentials.refresh_token,
                "expiry": credentials.expiry.isoformat()
            })

        return credentials

3. Google Drive API Integration​

3.1 Drive Service Wrapper​

class GoogleDriveService:
    """Google Drive API integration service."""

    def __init__(self, credentials: Credentials):
        self.service = build('drive', 'v3', credentials=credentials)

    # ==================== File Operations ====================

    async def create_file(
        self,
        name: str,
        content: bytes,
        mime_type: str,
        parent_folder_id: Optional[str] = None,
        properties: Optional[dict] = None
    ) -> DriveFile:
        """Create a new file in Google Drive."""

        file_metadata = {
            'name': name,
            'mimeType': mime_type,
        }

        if parent_folder_id:
            file_metadata['parents'] = [parent_folder_id]

        if properties:
            file_metadata['properties'] = properties  # Custom metadata

        media = MediaIoBaseUpload(
            io.BytesIO(content),
            mimetype=mime_type,
            resumable=True
        )

        file = self.service.files().create(
            body=file_metadata,
            media_body=media,
            fields='id, name, mimeType, webViewLink, version, createdTime'
        ).execute()

        return DriveFile(**file)

    async def get_file(self, file_id: str) -> DriveFile:
        """Get file metadata."""
        file = self.service.files().get(
            fileId=file_id,
            fields='id, name, mimeType, size, webViewLink, webContentLink, '
                   'version, createdTime, modifiedTime, owners, permissions'
        ).execute()
        return DriveFile(**file)

    async def update_file(
        self,
        file_id: str,
        content: Optional[bytes] = None,
        name: Optional[str] = None,
        properties: Optional[dict] = None
    ) -> DriveFile:
        """Update file content and/or metadata."""

        file_metadata = {}
        if name:
            file_metadata['name'] = name
        if properties:
            file_metadata['properties'] = properties

        media = None
        if content:
            media = MediaIoBaseUpload(
                io.BytesIO(content),
                mimetype='application/octet-stream',
                resumable=True
            )

        file = self.service.files().update(
            fileId=file_id,
            body=file_metadata if file_metadata else None,
            media_body=media,
            fields='id, name, version, modifiedTime'
        ).execute()

        return DriveFile(**file)

    async def delete_file(self, file_id: str) -> bool:
        """Delete a file (move to trash)."""
        self.service.files().delete(fileId=file_id).execute()
        return True

    # ==================== Folder Operations ====================

    async def create_folder(
        self,
        name: str,
        parent_id: Optional[str] = None
    ) -> DriveFile:
        """Create a folder in Google Drive."""

        file_metadata = {
            'name': name,
            'mimeType': 'application/vnd.google-apps.folder'
        }

        if parent_id:
            file_metadata['parents'] = [parent_id]

        folder = self.service.files().create(
            body=file_metadata,
            fields='id, name, webViewLink'
        ).execute()

        return DriveFile(**folder)

    # ==================== Search ====================

    async def search_files(
        self,
        query: str,
        folder_id: Optional[str] = None,
        mime_type: Optional[str] = None,
        page_size: int = 100,
        page_token: Optional[str] = None
    ) -> SearchResult:
        """Search files in Google Drive."""

        # Build query
        q_parts = []

        if query:
            q_parts.append(f"fullText contains '{query}'")

        if folder_id:
            q_parts.append(f"'{folder_id}' in parents")

        if mime_type:
            q_parts.append(f"mimeType = '{mime_type}'")

        q_parts.append("trashed = false")

        q = " and ".join(q_parts)

        results = self.service.files().list(
            q=q,
            pageSize=page_size,
            pageToken=page_token,
            fields='nextPageToken, files(id, name, mimeType, webViewLink, '
                   'modifiedTime, size, owners)',
            orderBy='modifiedTime desc'
        ).execute()

        return SearchResult(
            files=[DriveFile(**f) for f in results.get('files', [])],
            next_page_token=results.get('nextPageToken')
        )

    # ==================== Permissions ====================

    async def share_file(
        self,
        file_id: str,
        email: str,
        role: Literal["reader", "commenter", "writer", "owner"],
        send_notification: bool = True
    ) -> Permission:
        """Share file with a user."""

        permission = {
            'type': 'user',
            'role': role,
            'emailAddress': email
        }

        result = self.service.permissions().create(
            fileId=file_id,
            body=permission,
            sendNotificationEmail=send_notification,
            fields='id, type, role, emailAddress'
        ).execute()

        return Permission(**result)

    async def list_permissions(self, file_id: str) -> list[Permission]:
        """List all permissions on a file."""
        results = self.service.permissions().list(
            fileId=file_id,
            fields='permissions(id, type, role, emailAddress, displayName)'
        ).execute()

        return [Permission(**p) for p in results.get('permissions', [])]

    # ==================== Versions ====================

    async def list_revisions(self, file_id: str) -> list[Revision]:
        """List all revisions of a file."""
        results = self.service.revisions().list(
            fileId=file_id,
            fields='revisions(id, modifiedTime, lastModifyingUser, size)'
        ).execute()

        return [Revision(**r) for r in results.get('revisions', [])]

    async def get_revision(self, file_id: str, revision_id: str) -> bytes:
        """Download a specific revision of a file."""
        request = self.service.revisions().get_media(
            fileId=file_id,
            revisionId=revision_id
        )
        content = io.BytesIO()
        downloader = MediaIoBaseDownload(content, request)

        done = False
        while not done:
            _, done = downloader.next_chunk()

        return content.getvalue()

3.2 Shared Drives (Team Drives)​

class SharedDriveService:
    """Shared Drives management for team-owned content."""

    async def create_shared_drive(
        self,
        name: str,
        theme_id: Optional[str] = None
    ) -> SharedDrive:
        """Create a new Shared Drive."""

        request_id = str(uuid4())  # Required for idempotency

        drive_metadata = {
            'name': name
        }

        if theme_id:
            drive_metadata['themeId'] = theme_id

        drive = self.service.drives().create(
            requestId=request_id,
            body=drive_metadata,
            fields='id, name, createdTime'
        ).execute()

        return SharedDrive(**drive)

    async def add_member(
        self,
        drive_id: str,
        email: str,
        role: Literal["organizer", "fileOrganizer", "writer", "commenter", "reader"]
    ) -> Permission:
        """Add a member to a Shared Drive."""

        permission = {
            'type': 'user',
            'role': role,
            'emailAddress': email
        }

        result = self.service.permissions().create(
            fileId=drive_id,
            body=permission,
            supportsAllDrives=True,
            fields='id, type, role, emailAddress'
        ).execute()

        return Permission(**result)

    async def list_shared_drives(self) -> list[SharedDrive]:
        """List all Shared Drives accessible to the user."""
        results = self.service.drives().list(
            fields='drives(id, name, createdTime, restrictions)'
        ).execute()

        return [SharedDrive(**d) for d in results.get('drives', [])]

4. Google Docs API Integration​

4.1 Document Service​

class GoogleDocsService:
    """Google Docs API integration for document operations."""

    def __init__(self, credentials: Credentials):
        self.docs_service = build('docs', 'v1', credentials=credentials)
        self.drive_service = build('drive', 'v3', credentials=credentials)

    async def create_document(
        self,
        title: str,
        parent_folder_id: Optional[str] = None
    ) -> GoogleDoc:
        """Create a new Google Doc."""

        doc = self.docs_service.documents().create(
            body={'title': title}
        ).execute()

        # Move to folder if specified
        if parent_folder_id:
            self.drive_service.files().update(
                fileId=doc['documentId'],
                addParents=parent_folder_id,
                fields='id, parents'
            ).execute()

        return GoogleDoc(
            document_id=doc['documentId'],
            title=doc['title'],
            revision_id=doc['revisionId']
        )

    async def populate_template(
        self,
        template_id: str,
        replacements: dict[str, str],
        output_title: str,
        output_folder_id: Optional[str] = None
    ) -> GoogleDoc:
        """Copy template and replace placeholders."""

        # 1. Copy the template
        copy = self.drive_service.files().copy(
            fileId=template_id,
            body={'name': output_title}
        ).execute()

        doc_id = copy['id']

        # 2. Build replacement requests
        requests = []
        for placeholder, value in replacements.items():
            requests.append({
                'replaceAllText': {
                    'containsText': {
                        'text': '{{' + placeholder + '}}',
                        'matchCase': True
                    },
                    'replaceText': value
                }
            })

        # 3. Execute replacements
        if requests:
            self.docs_service.documents().batchUpdate(
                documentId=doc_id,
                body={'requests': requests}
            ).execute()

        # 4. Move to folder if specified
        if output_folder_id:
            self.drive_service.files().update(
                fileId=doc_id,
                addParents=output_folder_id,
                fields='id, parents'
            ).execute()

        return GoogleDoc(document_id=doc_id, title=output_title)

    async def insert_content(
        self,
        document_id: str,
        content: str,
        index: int = 1
    ) -> dict:
        """Insert content at a specific index in the document."""

        requests = [{
            'insertText': {
                'location': {'index': index},
                'text': content
            }
        }]

        result = self.docs_service.documents().batchUpdate(
            documentId=document_id,
            body={'requests': requests}
        ).execute()

        return result

    async def export_as_pdf(self, document_id: str) -> bytes:
        """Export Google Doc as PDF."""

        request = self.drive_service.files().export_media(
            fileId=document_id,
            mimeType='application/pdf'
        )

        content = io.BytesIO()
        downloader = MediaIoBaseDownload(content, request)

        done = False
        while not done:
            _, done = downloader.next_chunk()

        return content.getvalue()

    async def export_as_docx(self, document_id: str) -> bytes:
        """Export Google Doc as Microsoft Word (.docx)."""

        request = self.drive_service.files().export_media(
            fileId=document_id,
            mimeType='application/vnd.openxmlformats-officedocument.wordprocessingml.document'
        )

        content = io.BytesIO()
        downloader = MediaIoBaseDownload(content, request)

        done = False
        while not done:
            _, done = downloader.next_chunk()

        return content.getvalue()

4.2 Template Management​

class TemplateManager:
    """Manage document templates in Google Drive."""

    TEMPLATE_FOLDER_NAME = "CODITECT_Templates"

    async def list_templates(self) -> list[Template]:
        """List all available templates."""

        # Find template folder
        folder = await self.drive_service.search_files(
            query=f"name = '{self.TEMPLATE_FOLDER_NAME}'",
            mime_type='application/vnd.google-apps.folder'
        )

        if not folder.files:
            return []

        folder_id = folder.files[0].id

        # List templates in folder
        templates = await self.drive_service.search_files(
            folder_id=folder_id,
            mime_type='application/vnd.google-apps.document'
        )

        return [
            Template(
                id=f.id,
                name=f.name,
                preview_url=f.webViewLink,
                created_at=f.createdTime
            )
            for f in templates.files
        ]

    async def create_from_template(
        self,
        template_id: str,
        data: dict,
        output_name: str,
        output_folder_id: str
    ) -> GoogleDoc:
        """Create a new document from template with data merge."""

        return await self.docs_service.populate_template(
            template_id=template_id,
            replacements=data,
            output_title=output_name,
            output_folder_id=output_folder_id
        )

5. Google Cloud Storage Integration​

5.1 Backup Service​

class GCSBackupService:
    """Google Cloud Storage backup and archival service."""

    def __init__(self):
        self.client = storage.Client()
        self.bucket_name = os.environ["GCS_BACKUP_BUCKET"]
        self.bucket = self.client.bucket(self.bucket_name)

    async def backup_document(
        self,
        document_id: UUID,
        content: bytes,
        metadata: dict
    ) -> BackupResult:
        """Backup document to Cloud Storage with lifecycle management."""

        # Generate backup path with timestamp
        timestamp = datetime.utcnow().strftime("%Y/%m/%d/%H%M%S")
        blob_path = f"documents/{document_id}/{timestamp}/content"

        blob = self.bucket.blob(blob_path)

        # Set metadata
        blob.metadata = {
            "document_id": str(document_id),
            "backup_timestamp": datetime.utcnow().isoformat(),
            "original_name": metadata.get("name"),
            "content_hash": hashlib.sha256(content).hexdigest(),
            **metadata
        }

        # Upload with checksum verification
        blob.upload_from_string(
            content,
            content_type=metadata.get("mime_type", "application/octet-stream"),
            checksum="crc32c"
        )

        return BackupResult(
            blob_path=blob_path,
            size_bytes=len(content),
            checksum=blob.crc32c,
            backup_timestamp=datetime.utcnow()
        )

    async def restore_from_backup(
        self,
        document_id: UUID,
        backup_timestamp: Optional[datetime] = None
    ) -> bytes:
        """Restore document from backup."""

        if backup_timestamp:
            # Restore specific version
            prefix = f"documents/{document_id}/{backup_timestamp.strftime('%Y/%m/%d')}"
        else:
            # Get latest backup
            prefix = f"documents/{document_id}/"

        blobs = list(self.bucket.list_blobs(prefix=prefix))
        if not blobs:
            raise BackupNotFoundError(f"No backup found for {document_id}")

        # Get latest blob
        latest_blob = sorted(blobs, key=lambda b: b.time_created, reverse=True)[0]

        return latest_blob.download_as_bytes()

    async def list_backups(self, document_id: UUID) -> list[BackupInfo]:
        """List all backups for a document."""

        prefix = f"documents/{document_id}/"
        blobs = self.bucket.list_blobs(prefix=prefix)

        return [
            BackupInfo(
                blob_path=blob.name,
                timestamp=blob.time_created,
                size_bytes=blob.size,
                storage_class=blob.storage_class
            )
            for blob in blobs
        ]

5.2 Lifecycle Management​

class GCSLifecycleManager:
    """Manage storage lifecycle rules for cost optimization."""

    LIFECYCLE_RULES = [
        # Move to Nearline after 30 days
        {
            "action": {"type": "SetStorageClass", "storageClass": "NEARLINE"},
            "condition": {
                "age": 30,
                "matchesStorageClass": ["STANDARD"]
            }
        },
        # Move to Coldline after 365 days
        {
            "action": {"type": "SetStorageClass", "storageClass": "COLDLINE"},
            "condition": {
                "age": 365,
                "matchesStorageClass": ["NEARLINE"]
            }
        },
        # Move to Archive after 3 years
        {
            "action": {"type": "SetStorageClass", "storageClass": "ARCHIVE"},
            "condition": {
                "age": 1095,  # 3 years
                "matchesStorageClass": ["COLDLINE"]
            }
        },
        # Delete non-current versions after 90 days
        {
            "action": {"type": "Delete"},
            "condition": {
                "age": 90,
                "isLive": False
            }
        }
    ]

    async def apply_lifecycle_rules(self, bucket_name: str):
        """Apply lifecycle rules to bucket."""

        bucket = self.client.bucket(bucket_name)
        bucket.lifecycle_rules = self.LIFECYCLE_RULES
        bucket.patch()

        return {"status": "applied", "rules_count": len(self.LIFECYCLE_RULES)}

5.3 Bucket Lock for Compliance​

class ComplianceBucketLock:
    """Implement retention policies with Bucket Lock for compliance."""

    async def set_retention_policy(
        self,
        bucket_name: str,
        retention_days: int,
        lock: bool = False
    ):
        """Set retention policy on bucket."""

        bucket = self.client.bucket(bucket_name)
        bucket.retention_period = retention_days * 24 * 60 * 60  # Convert to seconds
        bucket.patch()

        if lock:
            # WARNING: This is irreversible!
            bucket.lock_retention_policy()

        return {
            "bucket": bucket_name,
            "retention_days": retention_days,
            "locked": lock
        }

    async def set_object_hold(
        self,
        bucket_name: str,
        blob_name: str,
        hold_type: Literal["event_based", "temporary"]
    ):
        """Place hold on specific object."""

        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(blob_name)

        if hold_type == "event_based":
            blob.event_based_hold = True
        else:
            blob.temporary_hold = True

        blob.patch()

        return {"blob": blob_name, "hold_type": hold_type}

6. Publishing to Google Drive​

6.1 Publishing Workflow​

class DrivePublishingService:
    """Publish documents to Google Drive with enterprise controls."""

    async def publish_document(
        self,
        document_id: UUID,
        target_folder_id: str,
        publish_options: PublishOptions
    ) -> PublishResult:
        """Publish document to Google Drive."""

        # 1. Get document from CODITECT
        document = await self.document_service.get(document_id)

        # 2. Validate publishing permissions
        await self.validate_publish_permissions(document, publish_options)

        # 3. Determine output format
        if publish_options.format == "google_doc":
            # Convert to Google Doc
            drive_file = await self.create_google_doc(document, target_folder_id)
        elif publish_options.format == "pdf":
            # Export as PDF
            content = await self.export_as_pdf(document)
            drive_file = await self.upload_file(
                content,
                f"{document.name}.pdf",
                "application/pdf",
                target_folder_id
            )
        else:
            # Upload original file
            drive_file = await self.upload_file(
                document.content,
                document.name,
                document.mime_type,
                target_folder_id
            )

        # 4. Apply permissions
        if publish_options.share_with:
            for recipient in publish_options.share_with:
                await self.drive_service.share_file(
                    drive_file.id,
                    recipient.email,
                    recipient.role
                )

        # 5. Apply access restrictions
        if publish_options.prevent_download:
            await self.set_copy_restrictions(drive_file.id)

        if publish_options.expiration_date:
            await self.set_expiration(drive_file.id, publish_options.expiration_date)

        # 6. Record publishing event
        await self.audit_service.log(
            event_type="document_published",
            document_id=document_id,
            target="google_drive",
            drive_file_id=drive_file.id
        )

        return PublishResult(
            document_id=document_id,
            drive_file_id=drive_file.id,
            web_view_link=drive_file.webViewLink,
            published_at=datetime.utcnow()
        )

    async def set_copy_restrictions(self, file_id: str):
        """Prevent copying, printing, downloading."""

        self.drive_service.service.files().update(
            fileId=file_id,
            body={
                'copyRequiresWriterPermission': True,
                'viewersCanCopyContent': False
            }
        ).execute()

    async def set_expiration(self, file_id: str, expiration: datetime):
        """Set expiration date on shared link."""

        # Note: This requires permission update, not file update
        permissions = await self.drive_service.list_permissions(file_id)

        for perm in permissions:
            if perm.role in ['reader', 'commenter', 'writer']:
                self.drive_service.service.permissions().update(
                    fileId=file_id,
                    permissionId=perm.id,
                    body={'expirationTime': expiration.isoformat()}
                ).execute()

6.2 Publishing Options Schema​

@dataclass
class PublishOptions:
    """Options for publishing documents to Google Drive."""

    # Output format
    format: Literal["original", "google_doc", "pdf"] = "original"

    # Sharing
    share_with: list[ShareRecipient] = field(default_factory=list)
    anyone_with_link: bool = False
    link_role: Literal["reader", "commenter", "writer"] = "reader"

    # Restrictions
    prevent_download: bool = False
    prevent_print: bool = False
    prevent_copy: bool = False

    # Expiration
    expiration_date: Optional[datetime] = None

    # Notifications
    send_notification: bool = True
    notification_message: Optional[str] = None

    # Versioning
    update_existing: bool = False
    existing_file_id: Optional[str] = None


@dataclass
class ShareRecipient:
    email: str
    role: Literal["reader", "commenter", "writer", "owner"]
    send_notification: bool = True

7. WebHooks and Real-Time Sync​

7.1 Drive Change Notifications​

class DriveWebhookService:
    """Handle Google Drive push notifications."""

    WEBHOOK_ENDPOINT = "/api/v1/webhooks/google-drive"

    async def setup_watch(
        self,
        file_id: str,
        expiration_minutes: int = 1440  # 24 hours max
    ) -> WatchChannel:
        """Set up push notifications for file changes."""

        channel_id = str(uuid4())
        expiration = datetime.utcnow() + timedelta(minutes=expiration_minutes)

        body = {
            'id': channel_id,
            'type': 'web_hook',
            'address': f"{os.environ['BASE_URL']}{self.WEBHOOK_ENDPOINT}",
            'expiration': int(expiration.timestamp() * 1000)
        }

        response = self.drive_service.service.files().watch(
            fileId=file_id,
            body=body
        ).execute()

        # Store channel for renewal
        await self.store_channel(
            channel_id=channel_id,
            file_id=file_id,
            resource_id=response['resourceId'],
            expiration=expiration
        )

        return WatchChannel(**response)

    async def handle_notification(
        self,
        headers: dict,
        body: bytes
    ) -> NotificationResult:
        """Handle incoming Drive push notification."""

        channel_id = headers.get('X-Goog-Channel-ID')
        resource_id = headers.get('X-Goog-Resource-ID')
        resource_state = headers.get('X-Goog-Resource-State')

        if resource_state == 'sync':
            # Initial sync confirmation
            return NotificationResult(action='sync_confirmed')

        if resource_state == 'change':
            # File changed - fetch details
            channel = await self.get_channel(channel_id)
            file = await self.drive_service.get_file(channel.file_id)

            # Sync changes to CODITECT
            await self.sync_changes(file)

            return NotificationResult(
                action='synced',
                file_id=file.id,
                modified_time=file.modifiedTime
            )

        return NotificationResult(action='ignored', state=resource_state)

8. API Endpoints​

8.1 Google Integration Endpoints​

# OAuth Flow
GET /api/v1/auth/google/authorize
  → Redirects to Google OAuth consent screen

GET /api/v1/auth/google/callback?code={auth_code}
  → Exchanges code for tokens, stores securely

POST /api/v1/auth/google/revoke
  → Revokes Google access tokens

# Drive Operations
GET /api/v1/google/drive/files?folder_id={id}&query={search}
POST /api/v1/google/drive/files
  Body: { name, content_base64, mime_type, folder_id }

GET /api/v1/google/drive/files/{file_id}
PUT /api/v1/google/drive/files/{file_id}
DELETE /api/v1/google/drive/files/{file_id}

# Shared Drives
GET /api/v1/google/drive/shared-drives
POST /api/v1/google/drive/shared-drives
  Body: { name }

# Publishing
POST /api/v1/documents/{id}/publish/google-drive
  Body: PublishOptions

# Templates
GET /api/v1/google/docs/templates
POST /api/v1/google/docs/from-template
  Body: { template_id, data, output_name, folder_id }

# Backups
POST /api/v1/documents/{id}/backup/gcs
GET /api/v1/documents/{id}/backups
POST /api/v1/documents/{id}/restore?backup_timestamp={ts}

# Webhooks
POST /api/v1/webhooks/google-drive
  (Receives push notifications from Google)

9. Security Considerations​

9.1 Token Security​

  • OAuth tokens encrypted at rest using AES-256
  • Refresh tokens stored in separate, more secure storage
  • Token rotation on each refresh
  • Automatic revocation on suspicious activity

9.2 Data in Transit​

  • All API calls over HTTPS (TLS 1.3)
  • Certificate pinning for Google API endpoints
  • Request signing for sensitive operations

9.3 Audit Logging​

  • All Google API calls logged with user context
  • Integration with Cloud Audit Logs for compliance
  • Real-time alerts on sensitive operations (share externally, delete)

10. Implementation Checklist​

Phase 1: Foundation (Weeks 1-2)​

  • Implement OAuth 2.0 flow with token management
  • Create Drive API wrapper service
  • Build basic file CRUD operations
  • Implement search functionality

Phase 2: Advanced Features (Weeks 3-4)​

  • Implement Shared Drives support
  • Build Google Docs API integration
  • Create template management system
  • Implement publishing workflow

Phase 3: Storage & Sync (Weeks 5-6)​

  • Build Cloud Storage backup service
  • Implement lifecycle management
  • Create webhook handlers for real-time sync
  • Build restore functionality

Phase 4: Enterprise Features (Weeks 7-8)​

  • Implement service account delegation
  • Build Admin SDK integration
  • Create compliance reporting
  • Implement audit log integration

Document Version: 1.0.0 Effective Date: Upon Approval Review Date: Quarterly Owner: Integration Team