# IMAP Connectivity Feature Design ## Overview This document outlines the design for implementing IMAP connectivity in the Email Organizer application. The feature will allow users to configure their IMAP server connection details, test the connection, and synchronize their email folders from the IMAP server to the application. ## Current State Analysis ### Existing Infrastructure - **User Model**: Already has `imap_config` field (JSON type) for storing IMAP configuration - **Authentication**: User authentication system is already implemented - **Folder Management**: Basic folder CRUD operations exist - **UI Framework**: Uses htmx, AlpineJS, and DaisyUI for dynamic interfaces - **Database**: PostgreSQL with SQLAlchemy ORM ### Gaps - No IMAP connection handling logic - No IMAP configuration UI - No folder synchronization from IMAP server - No IMAP testing functionality ## System Architecture ### High-Level Flow ```mermaid graph TD A[User Clicks IMAP Config] --> B[Show IMAP Configuration Modal] B --> C[User Enters Connection Details] C --> D{User Tests Connection} D -->|Success| E[Show Success Message] D -->|Failure| F[Show Error Message] E --> G[User Syncs Folders] F --> C G --> H[Fetch IMAP Folders] H --> I[Create Local Folders] I --> J[Update UI] ``` ### Data Flow ```mermaid sequenceDiagram participant U as User participant M as Main UI participant S as Server participant IMAP as IMAP Server participant DB as Database U->>M: Click "Configure IMAP" M->>S: GET /api/imap/config S->>M: Return IMAP config modal M->>U: Show configuration form U->>M: Enter connection details M->>S: POST /api/imap/test S->>IMAP: Test connection IMAP-->>S: Connection result S->>M: Return test result M->>U: Show test result U->>M: Click "Sync Folders" M->>S: POST /api/imap/sync S->>IMAP: List folders IMAP-->>S: Folder list S->>DB: Create/update folders DB-->>S: Success/failure S->>M: Return sync result M->>U: Show sync result ``` ## Database Schema ### User Model Updates The existing `User` model already has the `imap_config` field: ```python class User(Base, UserMixin): # ... existing fields ... imap_config = db.Column(db.JSON) # Store IMAP connection details # ... existing fields ... ``` ### Folder Model Updates The `Folder` model now includes a new field for storing recent email information: ```python class Folder(Base): # ... existing fields ... recent_emails = db.Column(db.JSON, default=list) # Store recent email subjects with dates # ... existing fields ... ``` ### IMAP Configuration Structure ```json { "server": "imap.gmail.com", "port": 993, "username": "user@gmail.com", "password": "app_password", "use_ssl": true, "use_tls": false, "connection_timeout": 30, "folder_prefix": "" } ``` ## IMAP Service Module ### File: `app/imap_service.py` ```python import imaplib import ssl import logging from typing import List, Dict, Optional, Tuple from email.header import decode_header from app.models import db, Folder, User from app import create_app class IMAPService: def __init__(self, user: User): self.user = user self.config = user.imap_config or {} self.connection = None def test_connection(self) -> Tuple[bool, str]: """Test IMAP connection with current configuration.""" try: if not self.config: return False, "No IMAP configuration found" # Create SSL context context = ssl.create_default_context() # Connect to IMAP server self.connection = imaplib.IMAP4_SSL( self.config.get('server', 'imap.gmail.com'), self.config.get('port', 993) ) # Login self.connection.login( self.config.get('username', ''), self.config.get('password', '') ) # Select inbox to verify connection self.connection.select('INBOX') # Close connection self.connection.close() self.connection.logout() return True, "Connection successful" except imaplib.IMAP4.error as e: return False, f"IMAP connection error: {str(e)}" except Exception as e: return False, f"Connection error: {str(e)}" finally: if self.connection: try: self.connection.logout() except: pass def get_folders(self) -> List[Dict[str, str]]: """Get list of folders from IMAP server.""" try: if not self.config: return [] # Create SSL context context = ssl.create_default_context() # Connect to IMAP server self.connection = imaplib.IMAP4_SSL( self.config.get('server', 'imap.gmail.com'), self.config.get('port', 993) ) # Login self.connection.login( self.config.get('username', ''), self.config.get('password', '') ) # List folders status, folder_data = self.connection.list() if status != 'OK': return [] folders = [] for folder_item in folder_data: if isinstance(folder_item, bytes): folder_item = folder_item.decode('utf-8') # Parse folder name (handle different IMAP server formats) parts = folder_item.split('"') if len(parts) >= 3: folder_name = parts[-1] if parts[-1] else parts[-2] else: folder_name = folder_item.split()[-1] # Handle nested folders (convert to slash notation) if folder_name.startswith('"') and folder_name.endswith('"'): folder_name = folder_name[1:-1] folders.append({ 'name': folder_name, 'full_path': folder_name }) # Close connection self.connection.close() self.connection.logout() return folders except Exception as e: logging.error(f"Error fetching IMAP folders: {str(e)}") return [] finally: if self.connection: try: self.connection.logout() except: pass def sync_folders(self) -> Tuple[bool, str]: """Sync IMAP folders with local database.""" try: if not self.config: return False, "No IMAP configuration found" # Get folders from IMAP server imap_folders = self.get_folders() if not imap_folders: return False, "No folders found on IMAP server" # Process each folder synced_count = 0 for imap_folder in imap_folders: folder_name = imap_folder['name'] # Skip special folders that might not be needed if folder_name.lower() in ['inbox', 'sent', 'drafts', 'spam', 'trash']: continue # Handle nested folder names (convert slashes to underscores or keep as-is) # According to requirements, nested folders should be created with slashes in the name display_name = folder_name # Check if folder already exists existing_folder = Folder.query.filter_by( user_id=self.user.id, name=display_name ).first() if not existing_folder: # Create new folder new_folder = Folder( user_id=self.user.id, name=display_name, rule_text=f"Auto-synced from IMAP folder: {folder_name}", priority=0 # Default priority ) db.session.add(new_folder) synced_count += 1 db.session.commit() return True, f"Successfully synced {synced_count} folders" except Exception as e: db.session.rollback() return False, f"Sync error: {str(e)}" def get_recent_emails(self, folder_name: str, limit: int = 3) -> List[Dict[str, any]]: """Get the most recent email subjects and dates from a specific folder.""" try: # Connect to IMAP server self._connect() # Login self.connection.login( self.config.get('username', ''), self.config.get('password', '') ) # Select the folder resp_code, content = self.connection.select(folder_name) if resp_code != 'OK': return [] # Get email IDs (most recent first) resp_code, content = self.connection.search(None, 'ALL') if resp_code != 'OK': return [] # Get the most recent emails (limit to 3) email_ids = content[0].split() recent_email_ids = email_ids[:limit] if len(email_ids) >= limit else email_ids recent_emails = [] for email_id in reversed(recent_email_ids): # Process from newest to oldest # Fetch the email headers resp_code, content = self.connection.fetch(email_id, '(RFC822.HEADER)') if resp_code != 'OK': continue # Parse the email headers raw_email = content[0][1] import email msg = email.message_from_bytes(raw_email) # Extract subject and date subject = msg.get('Subject', 'No Subject') date_str = msg.get('Date', '') # Decode the subject if needed try: decoded_parts = decode_header(subject) subject = ''.join([str(part, encoding or 'utf-8') if isinstance(part, bytes) else part for part, encoding in decoded_parts]) except Exception: pass # If decoding fails, use the original subject # Parse date if available try: email_date = parsedate_to_datetime(date_str) if date_str else None except Exception: email_date = None recent_emails.append({ 'subject': subject, 'date': email_date.isoformat() if email_date else None }) # Close folder and logout self.connection.close() self.connection.logout() self.connection = None return recent_emails except Exception as e: logging.error(f"Error getting recent emails for folder {folder_name}: {str(e)}") if self.connection: try: self.connection.logout() except: pass self.connection = None return [] ``` ## Routes Implementation ### File: `app/routes.py` (additions) ```python # Add to existing imports from app.imap_service import IMAPService import json @main.route('/api/imap/config', methods=['GET']) @login_required def imap_config_modal(): """Return the IMAP configuration modal.""" response = make_response(render_template('partials/imap_config_modal.html')) response.headers['HX-Trigger'] = 'open-modal' return response @main.route('/api/imap/test', methods=['POST']) @login_required def test_imap_connection(): """Test IMAP connection with provided configuration.""" try: # Get form data server = request.form.get('server') port = request.form.get('port') username = request.form.get('username') password = request.form.get('password') use_ssl = request.form.get('use_ssl') == 'on' # Validate required fields errors = {} if not server: errors['server'] = 'Server is required' if not port: errors['port'] = 'Port is required' elif not port.isdigit(): errors['port'] = 'Port must be a number' if not username: errors['username'] = 'Username is required' if not password: errors['password'] = 'Password is required' if errors: response = make_response(render_template('partials/imap_config_modal.html', errors=errors)) response.headers['HX-Retarget'] = '#imap-modal' response.headers['HX-Reswap'] = 'outerHTML' return response # Store configuration temporarily for testing test_config = { 'server': server, 'port': int(port), 'username': username, 'password': password, 'use_ssl': use_ssl, 'use_tls': False, 'connection_timeout': 30 } # Test connection temp_user = type('User', (), {'imap_config': test_config})() imap_service = IMAPService(temp_user) success, message = imap_service.test_connection() if success: # Save configuration to user's profile current_user.imap_config = test_config db.session.commit() response = make_response(render_template('partials/imap_config_modal.html', success=True, message=message)) response.headers['HX-Retarget'] = '#imap-modal' response.headers['HX-Reswap'] = 'outerHTML' else: response = make_response(render_template('partials/imap_config_modal.html', errors={'general': message})) response.headers['HX-Retarget'] = '#imap-modal' response.headers['HX-Reswap'] = 'outerHTML' return response except Exception as e: logging.exception("Error testing IMAP connection: %s", e) errors = {'general': 'An unexpected error occurred. Please try again.'} response = make_response(render_template('partials/imap_config_modal.html', errors=errors)) response.headers['HX-Retarget'] = '#imap-modal' response.headers['HX-Reswap'] = 'outerHTML' return response @main.route('/api/imap/sync', methods=['POST']) @login_required def sync_imap_folders(): """Sync folders from IMAP server.""" try: if not current_user.imap_config: return jsonify({'error': 'No IMAP configuration found. Please configure IMAP first.'}), 400 # Test connection first imap_service = IMAPService(current_user) success, message = imap_service.sync_folders() if success: # Get updated list of folders folders = Folder.query.filter_by(user_id=current_user.id).all() return render_template('partials/folders_list.html', folders=folders) else: return jsonify({'error': message}), 400 except Exception as e: logging.exception("Error syncing IMAP folders: %s", e) return jsonify({'error': 'An unexpected error occurred. Please try again.'}), 500 ``` ## UI Implementation ### Template: `app/templates/partials/imap_config_modal.html` ```html