integration-zoom-sdd-software-design-document

---

title: Zoom Integration - Software Design Document (SDD) type: reference version: 1.0.0 created: '2025-12-27' updated: '2025-12-27' status: active tags:

• ai-ml
• authentication
• deployment
• security
• testing
• api
• architecture
• automation summary: 'Zoom Integration - Software Design Document (SDD) Version: 2.0.0 Status: Draft Date: December 17, 2025 Author: CODITECT Architecture Team --- Executive Summary This document describes the software design for integrating Zoom video conferencing...'

---

Zoom Integration - Software Design Document (SDD)

Version: 2.0.0 Status: Draft Date: December 17, 2025 Author: CODITECT Architecture Team

---

1. Executive Summary

This document describes the software design for integrating Zoom video conferencing capabilities into the CODITECT platform. The integration provides meeting scheduling, participant management, and post-meeting artifact retrieval through Zoom's free-tier APIs.

1.1 Scope

• Meeting creation and management via Zoom REST API
• OAuth 2.0 authentication (User and Server-to-Server)
• Post-meeting transcript and recording retrieval
• CODITECT AI pipeline integration for meeting insights
• Optional component activation pattern

1.2 Design Principles

1. Zero External Cost - Only free-tier Zoom APIs
2. Provider Agnostic - Implements shared MeetingProviderInterface
3. Security First - Encrypted credential storage, minimal scopes
4. Resilient - Rate limiting, retries, graceful degradation

---

2. System Architecture

2.1 High-Level Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                    CODITECT Meeting Integration                     │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │              MeetingServiceOrchestrator                       │  │
│  │  - Provider selection and routing                             │  │
│  │  - Unified meeting operations                                 │  │
│  │  - Error handling and fallback                                │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                              │                                      │
│              ┌───────────────┼───────────────┐                      │
│              ▼               ▼               ▼                      │
│  ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐       │
│  │ GoogleMeet     │ │ ZoomProvider    │ │ TeamsProvider   │       │
│  │ Provider       │ │ (This Doc)      │ │ (Future)        │       │
│  └────────┬────────┘ └────────┬────────┘ └─────────────────┘       │
│           │                   │                                     │
└───────────┼───────────────────┼─────────────────────────────────────┘
            │                   │
            ▼                   ▼
    ┌───────────────┐   ┌───────────────┐
    │ Google APIs   │   │  Zoom APIs    │
    │ - Calendar    │   │  - REST API   │
    │ - Meet REST   │   │  - Webhooks   │
    │ - Drive       │   │  - OAuth 2.0  │
    └───────────────┘   └───────────────┘

2.2 Component Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                      ZoomProvider Component                         │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐  │
│  │   ZoomOAuth      │  │  ZoomMeetingAPI  │  │  ZoomWebhook     │  │
│  │   Manager        │  │  Client          │  │  Handler         │  │
│  │                  │  │                  │  │                  │  │
│  │ - User OAuth     │  │ - Create meeting │  │ - Event routing  │  │
│  │ - S2S OAuth      │  │ - Get meeting    │  │ - Recording ready│  │
│  │ - Token refresh  │  │ - List meetings  │  │ - Meeting ended  │  │
│  │ - Revocation     │  │ - Delete meeting │  │ - Transcript     │  │
│  └────────┬─────────┘  └────────┬─────────┘  └────────┬─────────┘  │
│           │                     │                     │            │
│           └──────────────┬──────┴─────────────────────┘            │
│                          ▼                                          │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │                    ZoomProvider                                │  │
│  │  - Implements MeetingProviderInterface                        │  │
│  │  - Coordinates auth, API calls, webhooks                      │  │
│  │  - Rate limiting and error handling                           │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                          │                                          │
│                          ▼                                          │
│  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐  │
│  │ ZoomRateLimiter  │  │ ZoomRecording    │  │ ZoomTranscript   │  │
│  │                  │  │ Client           │  │ Processor        │  │
│  │ - Token bucket   │  │                  │  │                  │  │
│  │ - Per-endpoint   │  │ - List recordings│  │ - Parse VTT      │  │
│  │ - Retry logic    │  │ - Get recording  │  │ - Format for AI  │  │
│  └──────────────────┘  │ - Download URL   │  │ - Diarization    │  │
│                        └──────────────────┘  └──────────────────┘  │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

---

3. Provider Interface Implementation

3.1 MeetingProviderInterface

The Zoom provider implements the shared abstract interface:

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List

@dataclass
class Meeting:
    id: str
    provider: str  # "zoom" | "google_meet" | "teams"
    title: str
    start_time: datetime
    end_time: Optional[datetime]
    join_url: str
    host_email: str
    participants: List['Participant']
    status: str  # "scheduled" | "started" | "ended"
    provider_data: dict  # Provider-specific metadata

@dataclass
class Transcript:
    meeting_id: str
    content: str
    entries: List['TranscriptEntry']
    language: str
    generated_at: datetime

@dataclass
class TranscriptEntry:
    speaker: str
    text: str
    start_time: float  # seconds
    end_time: float

@dataclass
class Recording:
    meeting_id: str
    recording_id: str
    download_url: str
    file_type: str  # "mp4" | "m4a" | "txt"
    file_size: int
    recording_start: datetime
    recording_end: datetime

@dataclass
class Participant:
    user_id: str
    name: str
    email: Optional[str]
    join_time: Optional[datetime]
    leave_time: Optional[datetime]
    duration_seconds: int

class MeetingProviderInterface(ABC):
    """Abstract interface for meeting providers."""

    @property
    @abstractmethod
    def provider_name(self) -> str:
        """Return provider identifier."""
        pass

    @abstractmethod
    async def authenticate(self, credentials: dict) -> bool:
        """Authenticate with the provider."""
        pass

    @abstractmethod
    async def create_meeting(
        self,
        title: str,
        start_time: datetime,
        duration_minutes: int,
        attendees: List[str] = None,
        **kwargs
    ) -> Meeting:
        """Create a new meeting."""
        pass

    @abstractmethod
    async def get_meeting(self, meeting_id: str) -> Optional[Meeting]:
        """Get meeting details by ID."""
        pass

    @abstractmethod
    async def list_meetings(
        self,
        start_date: datetime = None,
        end_date: datetime = None,
        limit: int = 100
    ) -> List[Meeting]:
        """List meetings within date range."""
        pass

    @abstractmethod
    async def delete_meeting(self, meeting_id: str) -> bool:
        """Delete/cancel a meeting."""
        pass

    @abstractmethod
    async def get_transcript(self, meeting_id: str) -> Optional[Transcript]:
        """Get meeting transcript if available."""
        pass

    @abstractmethod
    async def get_recording(self, meeting_id: str) -> Optional[Recording]:
        """Get meeting recording if available."""
        pass

    @abstractmethod
    async def get_participants(self, meeting_id: str) -> List[Participant]:
        """Get list of meeting participants."""
        pass

3.2 ZoomProvider Implementation

class ZoomProvider(MeetingProviderInterface):
    """Zoom implementation of meeting provider interface."""

    def __init__(self, config: 'ZoomConfig'):
        self.config = config
        self.auth_manager = ZoomOAuthManager(config)
        self.meeting_api = ZoomMeetingAPIClient(config)
        self.recording_client = ZoomRecordingClient(config)
        self.rate_limiter = ZoomRateLimiter(config)
        self.transcript_processor = ZoomTranscriptProcessor()

    @property
    def provider_name(self) -> str:
        return "zoom"

    async def authenticate(self, credentials: dict) -> bool:
        """Authenticate using OAuth 2.0 or Server-to-Server OAuth."""
        auth_type = credentials.get('type', 'oauth')

        if auth_type == 'oauth':
            return await self.auth_manager.authenticate_user(
                code=credentials.get('code'),
                redirect_uri=credentials.get('redirect_uri')
            )
        elif auth_type == 's2s':
            return await self.auth_manager.authenticate_s2s(
                account_id=credentials.get('account_id'),
                client_id=credentials.get('client_id'),
                client_secret=credentials.get('client_secret')
            )
        return False

    async def create_meeting(
        self,
        title: str,
        start_time: datetime,
        duration_minutes: int,
        attendees: List[str] = None,
        **kwargs
    ) -> Meeting:
        """Create a Zoom meeting."""
        await self.rate_limiter.acquire('meetings:write')

        meeting_data = await self.meeting_api.create_meeting(
            topic=title,
            start_time=start_time.isoformat(),
            duration=duration_minutes,
            type=2,  # Scheduled meeting
            settings={
                'host_video': kwargs.get('host_video', True),
                'participant_video': kwargs.get('participant_video', True),
                'join_before_host': kwargs.get('join_before_host', False),
                'mute_upon_entry': kwargs.get('mute_upon_entry', True),
                'auto_recording': kwargs.get('auto_recording', 'none'),
            }
        )

        return Meeting(
            id=str(meeting_data['id']),
            provider='zoom',
            title=meeting_data['topic'],
            start_time=datetime.fromisoformat(meeting_data['start_time'].rstrip('Z')),
            end_time=None,
            join_url=meeting_data['join_url'],
            host_email=meeting_data['host_email'],
            participants=[],
            status='scheduled',
            provider_data=meeting_data
        )

    async def get_transcript(self, meeting_id: str) -> Optional[Transcript]:
        """Get meeting transcript from cloud recording."""
        await self.rate_limiter.acquire('recordings:read')

        recordings = await self.recording_client.get_meeting_recordings(meeting_id)

        for file in recordings.get('recording_files', []):
            if file.get('file_type') == 'TRANSCRIPT':
                vtt_content = await self.recording_client.download_transcript(
                    file['download_url']
                )
                return self.transcript_processor.parse_vtt(
                    meeting_id=meeting_id,
                    vtt_content=vtt_content
                )

        return None

---

4. Authentication Design

4.1 OAuth 2.0 Flow (User-Level Access)

┌──────────┐     ┌──────────────┐     ┌────────────────┐
│   User   │     │   CODITECT   │     │   Zoom OAuth   │
└────┬─────┘     └──────┬───────┘     └───────┬────────┘
     │                  │                     │
     │  1. Request      │                     │
     │  Authorization   │                     │
     │─────────────────>│                     │
     │                  │                     │
     │                  │  2. Redirect to     │
     │                  │  Zoom Authorization │
     │                  │────────────────────>│
     │                  │                     │
     │  3. Login &      │                     │
     │  Consent         │                     │
     │────────────────────────────────────────>│
     │                  │                     │
     │                  │  4. Auth Code       │
     │                  │<────────────────────│
     │                  │                     │
     │                  │  5. Exchange for    │
     │                  │  Access Token       │
     │                  │────────────────────>│
     │                  │                     │
     │                  │  6. Access +        │
     │                  │  Refresh Token      │
     │                  │<────────────────────│
     │                  │                     │
     │  7. Success      │                     │
     │<─────────────────│                     │
     │                  │                     │

4.2 Server-to-Server OAuth (App-Level Access)

For automation without user intervention:

@dataclass
class ZoomS2SCredentials:
    account_id: str
    client_id: str
    client_secret: str

class ZoomOAuthManager:
    """Manages Zoom OAuth authentication."""

    OAUTH_TOKEN_URL = "https://zoom.us/oauth/token"

    async def authenticate_s2s(
        self,
        account_id: str,
        client_id: str,
        client_secret: str
    ) -> bool:
        """Authenticate using Server-to-Server OAuth."""

        auth_header = base64.b64encode(
            f"{client_id}:{client_secret}".encode()
        ).decode()

        async with aiohttp.ClientSession() as session:
            async with session.post(
                self.OAUTH_TOKEN_URL,
                headers={
                    'Authorization': f'Basic {auth_header}',
                    'Content-Type': 'application/x-www-form-urlencoded'
                },
                data={
                    'grant_type': 'account_credentials',
                    'account_id': account_id
                }
            ) as response:
                if response.status == 200:
                    token_data = await response.json()
                    self._store_token(token_data)
                    return True
                return False

4.3 Required OAuth Scopes

# Minimum required scopes
required_scopes:
  - meeting:read      # Read meeting details
  - meeting:write     # Create/update meetings
  - user:read         # Read user info

# Full functionality scopes
optional_scopes:
  - recording:read    # Access cloud recordings
  - report:read:admin # Access usage reports (admin)

---

5. API Integration Details

5.1 Zoom REST API Endpoints

Category	Endpoint	Method	Purpose
Meetings	/users/{userId}/meetings	POST	Create meeting
Meetings	/meetings/{meetingId}	GET	Get meeting details
Meetings	/meetings/{meetingId}	PATCH	Update meeting
Meetings	/meetings/{meetingId}	DELETE	Delete meeting
Meetings	/users/{userId}/meetings	GET	List user meetings
Recordings	/meetings/{meetingId}/recordings	GET	Get recordings
Participants	/past_meetings/{meetingId}/participants	GET	Get participants
Users	/users/me	GET	Get current user

5.2 Rate Limiting

Zoom API rate limits (free tier):

@dataclass
class ZoomRateLimits:
    """Zoom API rate limits by endpoint category."""

    # General API limits
    GENERAL_RATE_LIMIT = 100  # requests per second
    DAILY_LIMIT = 10000       # requests per day

    # Meeting API limits
    MEETING_CREATE = 100      # per day per user
    MEETING_READ = 30         # per second

    # Recording API limits
    RECORDING_READ = 30       # per second
    RECORDING_DOWNLOAD = 20   # per second

class ZoomRateLimiter:
    """Token bucket rate limiter for Zoom API."""

    def __init__(self, config: 'ZoomConfig'):
        self.limiters = {
            'meetings:read': TokenBucketLimiter(rate=30, burst=50),
            'meetings:write': TokenBucketLimiter(rate=10, burst=20),
            'recordings:read': TokenBucketLimiter(rate=30, burst=50),
            'recordings:download': TokenBucketLimiter(rate=20, burst=30),
            'users:read': TokenBucketLimiter(rate=30, burst=50),
        }

    async def acquire(self, operation: str) -> None:
        """Acquire permission to make an API call."""
        limiter = self.limiters.get(operation)
        if limiter:
            await limiter.acquire()

---

6. Data Models

6.1 Configuration Models

@dataclass
class ZoomConfig:
    """Configuration for Zoom integration."""

    # OAuth settings
    client_id: str
    client_secret: str
    redirect_uri: str

    # Server-to-Server OAuth (optional)
    account_id: Optional[str] = None

    # API settings
    api_base_url: str = "https://api.zoom.us/v2"
    webhook_secret_token: Optional[str] = None

    # Feature flags
    enable_recordings: bool = True
    enable_transcripts: bool = True
    enable_webhooks: bool = False

    # Rate limiting
    rate_limit_buffer: float = 0.8  # Use 80% of rate limit

@dataclass
class ZoomMeetingSettings:
    """Settings for creating Zoom meetings."""

    host_video: bool = True
    participant_video: bool = True
    join_before_host: bool = False
    mute_upon_entry: bool = True
    waiting_room: bool = True
    auto_recording: str = "none"  # "none" | "local" | "cloud"
    approval_type: int = 2  # 0=auto, 1=manual, 2=no registration

6.2 Response Models

@dataclass
class ZoomMeetingResponse:
    """Response from Zoom meeting creation."""

    id: int
    uuid: str
    topic: str
    type: int
    start_time: str
    duration: int
    timezone: str
    created_at: str
    join_url: str
    start_url: str
    password: str
    host_email: str
    settings: dict

@dataclass
class ZoomRecordingFile:
    """Individual recording file metadata."""

    id: str
    meeting_id: str
    recording_start: str
    recording_end: str
    file_type: str  # "MP4" | "M4A" | "TRANSCRIPT" | "CHAT"
    file_size: int
    download_url: str
    status: str  # "completed" | "processing"

---

7. Webhook Integration

7.1 Supported Webhook Events

Event	Trigger	Use Case
meeting.started	Meeting begins	Update status
meeting.ended	Meeting ends	Trigger post-processing
recording.completed	Recording ready	Download/process
recording.transcript_completed	Transcript ready	AI processing

7.2 Webhook Handler Design

class ZoomWebhookHandler:
    """Handles incoming Zoom webhooks."""

    def __init__(self, secret_token: str, event_handlers: dict):
        self.secret_token = secret_token
        self.event_handlers = event_handlers

    def verify_signature(self, payload: bytes, signature: str, timestamp: str) -> bool:
        """Verify webhook signature using HMAC-SHA256."""
        message = f"v0:{timestamp}:{payload.decode()}"
        expected = hmac.new(
            self.secret_token.encode(),
            message.encode(),
            hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(f"v0={expected}", signature)

    async def handle_event(self, event_type: str, payload: dict) -> None:
        """Route event to appropriate handler."""
        handler = self.event_handlers.get(event_type)
        if handler:
            await handler(payload)

# Event handlers
async def on_meeting_ended(payload: dict) -> None:
    """Handle meeting.ended event."""
    meeting_id = payload['payload']['object']['id']
    # Trigger post-meeting processing
    await process_meeting_artifacts(meeting_id)

async def on_recording_completed(payload: dict) -> None:
    """Handle recording.completed event."""
    meeting_id = payload['payload']['object']['id']
    recording_files = payload['payload']['object']['recording_files']
    # Download and process recordings
    await download_recordings(meeting_id, recording_files)

---

8. Error Handling

8.1 Exception Hierarchy

class ZoomIntegrationError(Exception):
    """Base exception for Zoom integration errors."""
    pass

class ZoomAuthenticationError(ZoomIntegrationError):
    """Authentication or authorization failure."""
    pass

class ZoomRateLimitError(ZoomIntegrationError):
    """Rate limit exceeded."""
    def __init__(self, retry_after: int):
        self.retry_after = retry_after
        super().__init__(f"Rate limit exceeded. Retry after {retry_after} seconds")

class ZoomAPIError(ZoomIntegrationError):
    """General API error."""
    def __init__(self, code: int, message: str):
        self.code = code
        super().__init__(f"Zoom API error {code}: {message}")

class ZoomResourceNotFoundError(ZoomIntegrationError):
    """Requested resource not found."""
    pass

class ZoomRecordingNotReadyError(ZoomIntegrationError):
    """Recording still processing."""
    def __init__(self, expected_ready: datetime = None):
        self.expected_ready = expected_ready
        super().__init__("Recording not yet available")

8.2 Retry Logic

class ZoomRetryHandler:
    """Handles retries with exponential backoff."""

    MAX_RETRIES = 3
    BASE_DELAY = 1.0
    MAX_DELAY = 30.0

    async def execute_with_retry(
        self,
        operation: Callable,
        *args,
        **kwargs
    ) -> Any:
        """Execute operation with automatic retries."""

        last_exception = None

        for attempt in range(self.MAX_RETRIES):
            try:
                return await operation(*args, **kwargs)

            except ZoomRateLimitError as e:
                delay = min(e.retry_after, self.MAX_DELAY)
                await asyncio.sleep(delay)
                last_exception = e

            except ZoomAPIError as e:
                if e.code in (500, 502, 503, 504):  # Server errors
                    delay = min(
                        self.BASE_DELAY * (2 ** attempt) + random.uniform(0, 1),
                        self.MAX_DELAY
                    )
                    await asyncio.sleep(delay)
                    last_exception = e
                else:
                    raise

        raise last_exception

---

9. CODITECT Integration

9.1 Component Activation

{
  "component": "zoom-integration",
  "type": "optional",
  "version": "1.0.0",
  "status": "disabled",
  "dependencies": [
    "coditect-core",
    "meeting-provider-interface"
  ],
  "configuration": {
    "requires_oauth": true,
    "requires_webhooks": false,
    "minimum_zoom_plan": "free"
  }
}

9.2 AI Pipeline Integration

class ZoomAIPipelineConnector:
    """Connects Zoom transcripts to CODITECT AI pipeline."""

    def __init__(self, ai_client: 'CODITECTAIClient'):
        self.ai_client = ai_client

    async def process_transcript(
        self,
        transcript: Transcript,
        analysis_types: List[str] = None
    ) -> dict:
        """Send transcript to AI for analysis."""

        analysis_types = analysis_types or [
            'summary',
            'action_items',
            'key_decisions',
            'follow_ups'
        ]

        results = {}

        for analysis_type in analysis_types:
            prompt = self._build_prompt(transcript, analysis_type)
            results[analysis_type] = await self.ai_client.analyze(
                content=transcript.content,
                prompt=prompt,
                context={'meeting_id': transcript.meeting_id}
            )

        return results

---

10. Security Considerations

10.1 Credential Storage

• OAuth tokens encrypted at rest (AES-256)
• Refresh tokens stored separately from access tokens
• Webhook secrets stored in environment variables
• No credentials in source code or logs

10.2 API Security

• All API calls over HTTPS
• Webhook signature verification required
• Token rotation on refresh
• Minimum required OAuth scopes

10.3 Data Protection

• Transcripts processed in-memory when possible
• Recordings not stored permanently
• Meeting metadata sanitized before storage
• PII handling per customer data policies

---

11. Appendix

A. Zoom API Documentation

• Zoom API Reference
• OAuth 2.0 Guide
• Webhooks Guide
• Server-to-Server OAuth

B. Related Documents

• ADR-001: Provider Abstraction Pattern
• ADR-002: Zero-Cost Architecture
• ADR-003: OAuth Authentication Strategy
• TDD: Technical Design Document

---

Document Control:

• Created: December 17, 2025
• Owner: CODITECT Architecture Team
• Review Cycle: Quarterly