Skip to content

Architecture

Jump to bottom
Dragon edited this page Dec 22, 2025 · 1 revision

🏗️ NaviDuck Architecture

Last updated: 12/22/2025

📋 Architecture Overview

NaviDuck is built as a modular, extensible CLI browser with a clear separation of concerns. The architecture follows a layered design pattern with components communicating through well-defined interfaces.

graph TB
    subgraph "User Interface Layer"
        UI[UIManager]
        CLI[Command Line Interface]
        UX[User Experience]
    end
    
    subgraph "Business Logic Layer"
        SM[Search Manager]
        AI[NavAI Assistant]
        PL[Page Loader]
        NM[Network Manager]
    end
    
    subgraph "Data Layer"
        BS[Browser State]
        DS[Data Storage]
        TM[Tor Manager]
    end
    
    subgraph "External Services"
        SE[Search Engines]
        WEB[Web Content]
        TOR[Tor Network]
    end
    
    CLI --> UI
    UI --> SM
    UI --> AI
    UI --> PL
    UI --> BS
    
    SM --> NM
    AI --> NM
    PL --> NM
    
    NM --> SE
    NM --> WEB
    NM --> TM
    
    TM --> TOR
    
    BS --> DS
    SM --> DS
    AI --> DS
    
    style UI fill:#e1f5fe
    style SM fill:#f3e5f5
    style AI fill:#e8f5e8
    style BS fill:#fff3e0
    style NM fill:#fce4ec
Loading

🧱 Core Architecture Principles

1. Modular Design

Each component is self-contained with clear interfaces:

  • Loose Coupling: Components communicate through interfaces, not direct dependencies
  • High Cohesion: Each component has a single, well-defined responsibility
  • Replaceable: Components can be swapped without affecting others

2. Layered Architecture

┌─────────────────────────────────┐
│        User Interface           │ ← Presentation Layer
├─────────────────────────────────┤
│      Business Logic             │ ← Application Layer
├─────────────────────────────────┤
│      Data Access / Services     │ ← Data Layer
├─────────────────────────────────┤
│  External APIs / Network        │ ← Infrastructure Layer
└─────────────────────────────────┘

3. Event-Driven Communication

Components communicate through events and callbacks:

  • Command Pattern: User actions as commands
  • Observer Pattern: State changes notify interested components
  • Strategy Pattern: Different algorithms for search, parsing, etc.

4. Fault Tolerance

  • Graceful Degradation: Features degrade gracefully when dependencies fail
  • Fallback Systems: Multiple fallback mechanisms for critical functions
  • Error Isolation: Failures in one component don't crash the system

🏛️ Component Architecture

1. BrowserState - The Central Nervous System

Responsibilities:

  • Manage application state and configuration
  • Handle data persistence
  • Coordinate between components
  • Maintain session information

Class Structure:

class BrowserState:
    def __init__(self):
        # Configuration
        self.use_emoji: bool
        self.icons: dict
        self.current_engine: str
        
        # Data Stores
        self.history: List[dict]
        self.bookmarks: List[dict]
        self.current_results: List[dict]
        
        # Session State
        self.current_page: str
        self.current_url: str
        self.current_title: str
        
        # Network
        self.tor_enabled: bool
        self.tor_process: Optional[Process]
        
        # File Paths
        self.data_file: str
        self.config_file: str
    
    def load_data(self) -> None:
        """Load user data from disk"""
    
    def save_data(self) -> None:
        """Save user data to disk"""
    
    def load_config(self) -> None:
        """Load configuration from disk"""
    
    def save_config(self) -> None:
        """Save configuration to disk"""

State Management Flow:

graph LR
    A[User Action] --> B[BrowserState.update]
    B --> C[State Change]
    C --> D[Notify Components]
    D --> E[UI Update]
    D --> F[Data Persistence]
    E --> G[User Feedback]
    F --> H[Disk Storage]
Loading

2. UIManager - The Presentation Layer

Responsibilities:

  • Handle all user interactions
  • Render UI components
  • Process user commands
  • Manage input/output

Key Design Patterns:

Command Pattern Implementation:

class CommandHandler:
    def __init__(self):
        self.commands = {
            's': self.handle_search,
            'search': self.handle_search,
            'ai': self.handle_ai,
            'go': self.handle_navigation,
            # ... more commands
        }
    
    def handle_command(self, cmd_line: str) -> bool:
        parts = cmd_line.strip().split()
        if not parts:
            return False
        
        cmd = parts[0].lower()
        handler = self.commands.get(cmd)
        
        if handler:
            return handler(parts[1:])
        else:
            # Try as direct query
            return self.handle_unknown(cmd_line)

Observer Pattern for State Updates:

class UIManager:
    def __init__(self, state: BrowserState):
        self.state = state
        self.state.add_observer(self)  # Subscribe to state changes
    
    def on_state_change(self, change_type: str, data: dict):
        """Called when BrowserState changes"""
        if change_type == 'search_complete':
            self.show_results(data['results'])
        elif change_type == 'page_loaded':
            self.show_page(data['page'])
        # ... more change types

UI Component Hierarchy:

UIManager
├── InputHandler (Command parsing)
├── ScreenRenderer (Display management)
├── ComponentFactory (UI element creation)
└── EventDispatcher (User event routing)

3. SearchManager - The Search Engine

Responsibilities:

  • Query multiple search engines
  • Parse and normalize results
  • Handle CAPTCHA and rate limiting
  • Implement fallback strategies

Search Pipeline Architecture:

class SearchPipeline:
    def search(self, query: str, engine: str = None) -> List[dict]:
        # 1. Query Validation
        validated = self.validate_query(query)
        
        # 2. Engine Selection
        engine = self.select_engine(engine, query)
        
        # 3. Request Construction
        request = self.build_request(query, engine)
        
        # 4. Network Request
        response = self.make_request(request)
        
        # 5. CAPTCHA Detection
        if self.detect_captcha(response):
            return self.handle_captcha(query, engine)
        
        # 6. Result Parsing
        results = self.parse_results(response, engine)
        
        # 7. Result Ranking
        ranked = self.rank_results(results, query)
        
        # 8. Caching
        self.cache_results(query, engine, ranked)
        
        return ranked

Multi-Engine Strategy Pattern:

class SearchEngineStrategy:
    def __init__(self):
        self.engines = {
            'ddg': DuckDuckGoStrategy(),
            'ddg_api': DuckDuckGoAPIStrategy(),
            'google': GoogleStrategy(),
            'wikipedia': WikipediaStrategy(),
            'brave': BraveStrategy(),
        }
    
    def search(self, query: str, engine: str) -> List[dict]:
        strategy = self.engines.get(engine)
        if not strategy:
            strategy = self.get_fallback_strategy()
        
        try:
            return strategy.execute(query)
        except SearchFailed:
            return self.try_alternative(query, engine)

CAPTCHA Handling System:

graph TD
    A[Start Search] --> B{Detect CAPTCHA}
    B -->|Yes| C[Log CAPTCHA Event]
    C --> D[Switch Engine]
    D --> E[Retry Search]
    E --> F{Success?}
    F -->|No| G[Next Fallback]
    F -->|Yes| H[Return Results]
    B -->|No| I[Parse Results]
    I --> H
    
    G --> D
Loading

4. NetworkManager - The Communication Hub

Responsibilities:

  • Handle all HTTP/HTTPS requests
  • Manage connection pooling
  • Implement retry logic
  • Handle proxies and Tor routing

Connection Management:

class ConnectionPool:
    def __init__(self, max_pool_size: int = 10):
        self.pools = {}
        self.max_pool_size = max_pool_size
    
    def get_connection(self, host: str) -> Connection:
        if host not in self.pools:
            self.pools[host] = ConnectionPoolForHost(host, self.max_pool_size)
        
        return self.pools[host].get_connection()
    
    def release_connection(self, host: str, conn: Connection):
        self.pools[host].release_connection(conn)

Request Pipeline with Middleware:

class RequestPipeline:
    def __init__(self):
        self.middleware = [
            UserAgentMiddleware(),
            RetryMiddleware(max_retries=3),
            TimeoutMiddleware(timeout=10),
            GzipMiddleware(),
            CookieMiddleware(),
            CacheMiddleware(),
            TorMiddleware(),  # Conditional
        ]
    
    def execute(self, request: Request) -> Response:
        response = request
        
        for middleware in self.middleware:
            if middleware.should_process(request):
                response = middleware.process(response)
        
        return response

5. NavAI - The Intelligent Assistant

Responsibilities:

  • Process natural language queries
  • Integrate with search APIs
  • Maintain conversation context
  • Provide intelligent responses

AI Processing Pipeline:

class AIPipeline:
    def process_query(self, query: str) -> str:
        # 1. Query Classification
        query_type = self.classify_query(query)
        
        # 2. Intent Recognition
        intent = self.extract_intent(query)
        
        # 3. Context Integration
        context = self.get_context()
        
        # 4. Knowledge Source Selection
        sources = self.select_sources(query_type, intent)
        
        # 5. Parallel Knowledge Gathering
        knowledge = self.gather_knowledge(sources, query, context)
        
        # 6. Response Generation
        response = self.generate_response(knowledge, query_type)
        
        # 7. Context Update
        self.update_context(query, response)
        
        return response

Knowledge Integration Architecture:

NavAI Knowledge System
├── Local Knowledge Base (Pre-defined responses)
├── DuckDuckGo Instant Answer API
├── Web Search Integration
├── Conversation Memory
└── Response Templates

6. TorManager - The Privacy Layer

Responsibilities:

  • Manage Tor process lifecycle
  • Handle SOCKS5 proxy configuration
  • Monitor Tor circuit health
  • Implement bridge support

Tor Process Management:

class TorProcessManager:
    def __init__(self):
        self.process = None
        self.data_dir = None
        self.port = 9050
    
    def start(self) -> bool:
        # 1. Create temp directory
        self.data_dir = tempfile.mkdtemp()
        
        # 2. Build command
        cmd = self.build_tor_command()
        
        # 3. Start process
        self.process = subprocess.Popen(cmd, ...)
        
        # 4. Wait for bootstrap
        return self.wait_for_bootstrap()
    
    def build_tor_command(self) -> List[str]:
        return [
            self.tor_exe,
            "--SocksPort", str(self.port),
            "--DataDirectory", self.data_dir,
            "--Log", "notice stdout",
            "--AvoidDiskWrites", "1",
        ]

Circuit Management:

class TorCircuitManager:
    def __init__(self):
        self.circuits = {}
        self.current_circuit = None
    
    def new_circuit(self) -> str:
        """Create new Tor circuit"""
        circuit_id = self.generate_circuit_id()
        
        self.circuits[circuit_id] = {
            'created': time.time(),
            'requests': 0,
            'bytes_sent': 0,
            'bytes_received': 0,
        }
        
        self.current_circuit = circuit_id
        return circuit_id
    
    def should_rotate(self, circuit_id: str) -> bool:
        """Determine if circuit should be rotated"""
        circuit = self.circuits.get(circuit_id)
        if not circuit:
            return True
        
        # Rotate based on age or usage
        age = time.time() - circuit['created']
        return age > 600 or circuit['requests'] > 100  # 10 min or 100 requests

🔄 Data Flow Architecture

Search Flow:

sequenceDiagram
    participant U as User
    participant UI as UIManager
    participant BS as BrowserState
    participant SM as SearchManager
    participant NM as NetworkManager
    participant SE as Search Engine
    participant DS as Data Storage
    
    U->>UI: Enter search query
    UI->>BS: Validate query
    BS->>UI: Return validation result
    UI->>SM: Execute search
    SM->>NM: Make HTTP request
    NM->>SE: Send search request
    SE->>NM: Return HTML/JSON
    NM->>SM: Pass response
    SM->>SM: Parse results
    SM->>BS: Store results
    BS->>DS: Persist to disk
    BS->>UI: Notify completion
    UI->>U: Display results
Loading

AI Query Flow:

sequenceDiagram
    participant U as User
    participant UI as UIManager
    participant AI as NavAI
    participant KB as Knowledge Base
    participant DDG as DuckDuckGo API
    participant SM as SearchManager
    
    U->>UI: Ask AI question
    UI->>AI: Process query
    AI->>KB: Check local knowledge
    KB->>AI: Return local answer
    alt Local answer found
        AI->>UI: Return local answer
    else No local answer
        AI->>DDG: Query Instant Answer API
        DDG->>AI: Return API answer
        alt API answer valid
            AI->>UI: Return API answer
        else No API answer
            AI->>SM: Fallback to search
            SM->>AI: Return search results
            AI->>UI: Return search-based answer
        end
    end
    UI->>U: Display answer
Loading

Page Loading Flow:

sequenceDiagram
    participant U as User
    participant UI as UIManager
    participant PL as PageLoader
    participant NM as NetworkManager
    participant TM as TorManager
    participant WEB as Website
    participant BS as BrowserState
    
    U->>UI: Request URL
    UI->>PL: Load page
    PL->>BS: Check Tor requirement
    alt .onion or Tor enabled
        BS->>TM: Get Tor proxy
        TM->>PL: Return proxy config
        PL->>NM: Request via Tor
    else Regular site
        PL->>NM: Direct request
    end
    NM->>WEB: HTTP GET
    WEB->>NM: Return content
    NM->>PL: Pass response
    PL->>PL: Extract content
    PL->>BS: Store page data
    BS->>UI: Notify completion
    UI->>U: Display page
Loading

🗄️ Data Architecture

Storage Schema:

1. Configuration Storage (~/.naviduck_config.json):

{
  "version": "1.0",
  "default_engine": "brave",
  "use_emoji": false,
  "tor_enabled": false,
  "user_agent": "Mozilla/5.0...",
  "engines": {
    "ddg": true,
    "ddg_api": true,
    "google": false,
    "wikipedia": true,
    "brave": true
  },
  "privacy": {
    "clear_history_on_exit": false,
    "strip_tracking_params": true,
    "randomize_user_agent": false
  },
  "performance": {
    "cache_enabled": true,
    "cache_ttl": 300,
    "timeout": 10
  }
}

2. User Data Storage (~/.naviduck_data.json):

{
  "history": [
    {
      "id": "timestamp_hash",
      "type": "search|visit|ai",
      "timestamp": "2024-01-15T10:30:00",
      "data": {
        "query": "python programming",
        "engine": "brave",
        "results_count": 10
      }
    }
  ],
  "bookmarks": [
    {
      "id": "url_hash",
      "title": "Python Official Website",
      "url": "https://python.org",
      "added": "2024-01-15T10:30:00",
      "tags": ["programming", "python"],
      "notes": "Main Python website"
    }
  ],
  "sessions": {
    "last_session": {
      "timestamp": "2024-01-15T10:30:00",
      "queries": ["python", "flask"],
      "visited": ["https://python.org"]
    }
  }
}

3. Cache Storage (~/.naviduck_cache/):

~/.naviduck_cache/
├── search/
│   ├── brave/
│   │   ├── python_123abc.cache
│   │   └── flask_456def.cache
│   └── ddg_api/
│       └── ai_query_789ghi.cache
├── pages/
│   └── https_python.org_index.html.cache
└── ai/
    └── responses/
        └── what_is_python.cache

Data Access Patterns:

Repository Pattern Implementation:

class DataRepository:
    def __init__(self, storage_backend):
        self.storage = storage_backend
    
    def save_history(self, entry: dict) -> bool:
        """Save history entry with validation"""
        validated = self.validate_history_entry(entry)
        return self.storage.append('history', validated)
    
    def get_recent_history(self, limit: int = 50) -> List[dict]:
        """Get recent history entries"""
        return self.storage.get_slice('history', -limit)
    
    def clear_old_history(self, max_age_days: int) -> int:
        """Clear history older than specified days"""
        cutoff = datetime.now() - timedelta(days=max_age_days)
        return self.storage.delete_by_condition(
            'history', 
            lambda x: datetime.fromisoformat(x['timestamp']) < cutoff
        )

Caching Strategy:

class MultiLevelCache:
    def __init__(self):
        self.memory_cache = {}  # LRU in-memory cache
        self.disk_cache = DiskCache()  # Persistent disk cache
    
    def get(self, key: str):
        # 1. Try memory cache
        if key in self.memory_cache:
            entry = self.memory_cache[key]
            if not self.is_expired(entry):
                self.update_lru(key)
                return entry['data']
        
        # 2. Try disk cache
        disk_data = self.disk_cache.get(key)
        if disk_data and not self.is_expired(disk_data):
            # Promote to memory cache
            self.memory_cache[key] = disk_data
            return disk_data['data']
        
        # 3. Cache miss
        return None
    
    def set(self, key: str, data, ttl: int = 300):
        entry = {
            'data': data,
            'timestamp': time.time(),
            'ttl': ttl
        }
        
        # Update both caches
        self.memory_cache[key] = entry
        self.disk_cache.set(key, entry)
        
        # Enforce memory cache limits
        if len(self.memory_cache) > self.max_memory_entries:
            self.evict_oldest()

🔌 Plugin & Extension Architecture

Plugin System Design:

Plugin Interface:

class NaviDuckPlugin:
    """Base class for all plugins"""
    
    def __init__(self, context):
        self.context = context  # Access to NaviDuck APIs
        self.name = "Unnamed Plugin"
        self.version = "1.0"
        self.description = ""
    
    def on_load(self) -> bool:
        """Called when plugin is loaded"""
        return True
    
    def on_unload(self) -> None:
        """Called when plugin is unloaded"""
        pass
    
    def on_command(self, command: str, args: List[str]) -> Optional[bool]:
        """Handle custom commands"""
        return None  # Return True if handled
    
    def on_search(self, query: str, engine: str, results: List[dict]) -> List[dict]:
        """Modify search results"""
        return results
    
    def on_page_load(self, url: str, content: str) -> str:
        """Modify page content"""
        return content
    
    def register_commands(self) -> Dict[str, Callable]:
        """Register custom commands"""
        return {}

Example Plugin:

class DarkReaderPlugin(NaviDuckPlugin):
    def __init__(self, context):
        super().__init__(context)
        self.name = "Dark Reader"
        self.version = "1.0"
        self.description = "Dark mode for web pages"
    
    def on_page_load(self, url: str, content: str) -> str:
        """Apply dark theme to HTML"""
        if not self.is_html(content):
            return content
        
        # Simple dark theme CSS injection
        dark_css = """
        <style>
        body { background: #1a1a1a; color: #e0e0e0; }
        a { color: #80c0ff; }
        </style>
        """
        
        # Inject CSS into HTML
        return self.inject_css(content, dark_css)

Extension Points:

1. Search Engine Plugins:

class SearchEnginePlugin(NaviDuckPlugin):
    def get_engine_config(self) -> dict:
        return {
            "id": "mysearch",
            "name": "My Search Engine",
            "url": "https://api.example.com/search",
            "parser": self.parse_results,
            "icon": "SEARCH"
        }

2. Content Filter Plugins:

class ContentFilterPlugin(NaviDuckPlugin):
    def __init__(self, context):
        super().__init__(context)
        self.filters = [
            self.remove_ads,
            self.clean_tracking,
            self.simplify_layout
        ]
    
    def on_page_load(self, url: str, content: str) -> str:
        for filter_func in self.filters:
            content = filter_func(content)
        return content

3. UI Theme Plugins:

class UIThemePlugin(NaviDuckPlugin):
    def get_theme(self) -> dict:
        return {
            "colors": {
                "PROMPT": "\033[1;35m",
                "SUCCESS": "\033[1;32m",
                "ERROR": "\033[1;31m"
            },
            "icons": {
                "SEARCH": "🔎",
                "AI": "🤖",
                "BOOKMARK": "📑"
            }
        }

🛡️ Security Architecture

Defense in Depth Strategy:

1. Input Validation Layer:

class SecurityValidator:
    def validate_url(self, url: str) -> bool:
        # Parse and validate URL
        parsed = urlparse(url)
        
        # Block dangerous schemes
        if parsed.scheme not in ['http', 'https']:
            return False
        
        # Block localhost/internal addresses
        if self.is_local_address(parsed.netloc):
            return False
        
        # Block known malicious patterns
        if self.contains_malicious_patterns(url):
            return False
        
        return True
    
    def sanitize_query(self, query: str) -> str:
        """Sanitize search queries"""
        # Remove control characters
        sanitized = ''.join(char for char in query if ord(char) >= 32)
        
        # Limit length
        if len(sanitized) > 500:
            sanitized = sanitized[:500]
        
        return sanitized

2. Network Security Layer:

class NetworkSecurity:
    def __init__(self):
        self.ssl_context = self.create_secure_ssl_context()
        self.request_validator = RequestValidator()
    
    def create_secure_ssl_context(self) -> ssl.SSLContext:
        context = ssl.create_default_context()
        context.minimum_version = ssl.TLSVersion.TLSv1_2
        context.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20')
        return context
    
    def make_secure_request(self, url: str) -> Response:
        if not self.request_validator.is_allowed(url):
            raise SecurityError("URL not allowed")
        
        # Use secure SSL context
        response = requests.get(
            url, 
            timeout=10,
            verify=self.ssl_context
        )
        
        # Validate response
        self.validate_response(response)
        
        return response

3. Data Protection Layer:

class DataProtection:
    def __init__(self, encryption_key: bytes):
        self.cipher = Fernet(encryption_key)
    
    def encrypt_data(self, data: dict) -> bytes:
        """Encrypt sensitive data"""
        json_data = json.dumps(data).encode()
        return self.cipher.encrypt(json_data)
    
    def decrypt_data(self, encrypted: bytes) -> dict:
        """Decrypt sensitive data"""
        decrypted = self.cipher.decrypt(encrypted)
        return json.loads(decrypted)
    
    def secure_delete(self, filepath: str) -> None:
        """Securely delete file by overwriting"""
        with open(filepath, 'rb+') as f:
            length = f.tell()
            f.seek(0)
            f.write(os.urandom(length))
        os.remove(filepath)

Threat Mitigation Matrix:

Threat Layer Mitigation
SQL Injection Input Validation Parameter sanitization
XSS Output Encoding HTML entity encoding
CSRF Session Management Token validation
MITM Network Security SSL/TLS, certificate pinning
Data Theft Data Protection Encryption at rest
DoS Rate Limiting Request throttling
Fingerprinting Privacy Layer Randomization, Tor

📈 Performance Architecture

Caching Strategy:

Multi-Level Cache Hierarchy:

┌─────────────────────────────────┐
│      Memory Cache (LRU)         │ ← Fastest, 1000 entries
├─────────────────────────────────┤
│      Disk Cache (SSD/HDD)       │ ← Persistent, 10k entries
├─────────────────────────────────┤
│      CDN/Edge Cache             │ ← External, distributed
├─────────────────────────────────┤
│      Origin Server              │ ← Slowest, always fresh
└─────────────────────────────────┘

Cache Implementation:

class PerformanceOptimizer:
    def __init__(self):
        self.caches = {
            'search': LRUCache(maxsize=100),
            'pages': LRUCache(maxsize=50),
            'ai': LRUCache(maxsize=200),
        }
        
        self.metrics = {
            'cache_hits': 0,
            'cache_misses': 0,
            'avg_response_time': 0,
        }
    
    def cached_search(self, query: str, engine: str) -> List[dict]:
        cache_key = f"{engine}:{query}"
        
        # Check cache
        cached = self.caches['search'].get(cache_key)
        if cached and not self.is_stale(cached):
            self.metrics['cache_hits'] += 1
            return cached
        
        # Cache miss - perform actual search
        self.metrics['cache_misses'] += 1
        start_time = time.time()
        
        results = self.perform_search(query, engine)
        
        # Calculate response time
        response_time = time.time() - start_time
        self.update_avg_time(response_time)
        
        # Cache results
        self.caches['search'].set(cache_key, results, ttl=300)
        
        return results

Connection Pooling:

class ConnectionPoolManager:
    def __init__(self, max_pool_size: int = 10):
        self.pools = {}
        self.max_pool_size = max_pool_size
        self.stats = defaultdict(int)
    
    def get_connection(self, host: str) -> Connection:
        if host not in self.pools:
            self.pools[host] = ConnectionPool(
                host=host,
                max_size=self.max_pool_size
            )
        
        conn = self.pools[host].get_connection()
        self.stats['connections_used'] += 1
        
        return conn
    
    def release_connection(self, host: str, conn: Connection):
        self.pools[host].release_connection(conn)
        self.stats['connections_released'] += 1

Lazy Loading:

class LazyLoader:
    def __init__(self, loader_func):
        self.loader_func = loader_func
        self._value = None
        self._loaded = False
    
    @property
    def value(self):
        if not self._loaded:
            self._value = self.loader_func()
            self._loaded = True
        return self._value
    
    def invalidate(self):
        self._loaded = False
        self._value = None

🔄 Error Handling Architecture

Error Hierarchy:

class NaviDuckError(Exception):
    """Base exception for all NaviDuck errors"""
    pass

class NetworkError(NaviDuckError):
    """Network-related errors"""
    pass

class SearchError(NaviDuckError):
    """Search-related errors"""
    pass

class AIError(NaviDuckError):
    """AI-related errors"""
    pass

class ConfigurationError(NaviDuckError):
    """Configuration-related errors"""
    pass

class SecurityError(NaviDuckError):
    """Security-related errors"""
    pass

Error Recovery Strategy:

class ErrorRecovery:
    def __init__(self):
        self.recovery_strategies = {
            NetworkError: self.recover_from_network_error,
            SearchError: self.recover_from_search_error,
            AIError: self.recover_from_ai_error,
        }
    
    def handle_error(self, error: Exception, context: dict) -> Any:
        """Handle error with appropriate recovery strategy"""
        error_type = type(error)
        
        if error_type in self.recovery_strategies:
            return self.recovery_strategies[error_type](error, context)
        
        # Default recovery
        return self.default_recovery(error, context)
    
    def recover_from_network_error(self, error: NetworkError, context: dict):
        # 1. Retry with exponential backoff
        for attempt in range(3):
            try:
                return self.retry_operation(context)
            except:
                time.sleep(2 ** attempt)  # Exponential backoff
        
        # 2. Switch to alternative endpoint
        return self.use_alternative_endpoint(context)
    
    def recover_from_search_error(self, error: SearchError, context: dict):
        # 1. Switch search engine
        alternative_engine = self.get_alternative_engine()
        return self.retry_with_engine(context, alternative_engine)

Circuit Breaker Pattern:

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    def execute(self, operation: Callable) -> Any:
        if self.state == "OPEN":
            if self.should_try_reset():
                self.state = "HALF_OPEN"
            else:
                raise CircuitBreakerOpen("Circuit breaker is open")
        
        try:
            result = operation()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        
        if self.failures >= self.failure_threshold:
            self.state = "OPEN"
    
    def on_success(self):
        self.failures = 0
        self.state = "CLOSED"

🧪 Testing Architecture

Test Pyramid:

        ┌─────────────────────┐
        │    E2E Tests        │ ← 10% of tests
        │  (Full system)      │
        ├─────────────────────┤
        │  Integration Tests  │ ← 20% of tests
        │ (Component interaction)│
        ├─────────────────────┤
        │    Unit Tests       │ ← 70% of tests
        │  (Individual units) │
        └─────────────────────┘

Test Suite Organization:

# tests/
# ├── unit/
# │   ├── test_browser_state.py
# │   ├── test_search_manager.py
# │   ├── test_network_manager.py
# │   └── test_ui_manager.py
# ├── integration/
# │   ├── test_search_flow.py
# │   ├── test_ai_flow.py
# │   └── test_tor_integration.py
# ├── e2e/
# │   ├── test_full_search.py
# │   └── test_user_journey.py
# └── conftest.py

Mocking Strategy:

class MockNetworkManager(NetworkManager):
    def __init__(self):
        self.responses = {}
        self.requests = []
    
    def add_mock_response(self, url: str, response: dict):
        self.responses[url] = response
    
    def get(self, url: str, **kwargs) -> Response:
        self.requests.append({
            'url': url,
            'timestamp': time.time(),
            'kwargs': kwargs
        })
        
        if url in self.responses:
            return MockResponse(self.responses[url])
        
        # Default mock response
        return MockResponse({
            'status_code': 200,
            'text': '<html>Mock response</html>',
            'headers': {'Content-Type': 'text/html'}
        })

🚀 Deployment & Build Architecture

Build Pipeline:

# .github/workflows/build.yml
name: Build and Test

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: [3.8, 3.9, 3.10, 3.11]
    
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: ${{ matrix.python-version }}
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest pytest-cov
    
    - name: Run tests
      run: |
        pytest --cov=naviduck tests/
    
    - name: Upload coverage
      uses: codecov/codecov-action@v2

Packaging Strategy:

# setup.py
from setuptools import setup, find_packages

setup(
    name="naviduck",
    version="1.0.0",
    packages=find_packages(),
    install_requires=[
        "requests>=2.25.0",
    ],
    extras_require={
        'dev': [
            'pytest>=6.0',
            'pytest-cov>=2.0',
            'black>=21.0',
            'flake8>=3.9',
        ],
        'tor': [
            'stem>=1.8.0',
        ],
    },
    entry_points={
        'console_scripts': [
            'naviduck=naviduck.main:main',
        ],
    },
    classifiers=[
        "Programming Language :: Python :: 3",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",
    ],
)

📊 Monitoring & Observability

Metrics Collection:

class MetricsCollector:
    def __init__(self):
        self.metrics = {
            'search': {
                'count': 0,
                'success': 0,
                'failure': 0,
                'avg_time': 0,
            },
            'ai': {
                'queries': 0,
                'api_calls': 0,
                'cache_hits': 0,
            },
            'network': {
                'requests': 0,
                'bytes_sent': 0,
                'bytes_received': 0,
            }
        }
    
    def record_search(self, success: bool, duration: float):
        self.metrics['search']['count'] += 1
        if success:
            self.metrics['search']['success'] += 1
        else:
            self.metrics['search']['failure'] += 1
        
        # Update average (moving average)
        current_avg = self.metrics['search']['avg_time']
        n = self.metrics['search']['count']
        self.metrics['search']['avg_time'] = (
            current_avg * (n-1) + duration
        ) / n
    
    def get_report(self) -> dict:
        return {
            'timestamp': time.time(),
            'metrics': self.metrics,
            'summary': self.generate_summary()
        }

Logging Architecture:

class StructuredLogger:
    def __init__(self):
        self.loggers = {}
        
    def get_logger(self, name: str):
        if name not in self.loggers:
            self.loggers[name] = Logger(name)
        return self.loggers[name]
    
    def log_event(self, event_type: str, data: dict):
        log_entry = {
            'timestamp': time.time(),
            'event': event_type,
            'data': data,
            'context': self.get_context()
        }
        
        # Write to structured log file
        self.write_log_entry(log_entry)
        
        # Also output to console in dev mode
        if self.is_dev_mode():
            print(f"[{event_type}] {json.dumps(data)}")

🔮 Future Architecture Evolution

Planned Architectural Improvements:

1. Microservices Architecture (v2.0):

┌─────────────────────────────────────────────┐
│                API Gateway                  │
├─────────────────────────────────────────────┤
│  ┌─────────┐ ┌─────────┐ ┌─────────┐      │
│  │ Search  │ │   AI    │ │  Proxy  │      │
│  │ Service │ │ Service │ │ Service │      │
│  └─────────┘ └─────────┘ └─────────┘      │
└─────────────────────────────────────────────┘

2. Plugin Architecture (v2.1):

  • Dynamic plugin loading
  • Plugin marketplace
  • Sandboxed plugin execution
  • Versioned plugin API

3. Distributed Architecture (v3.0):

  • Peer-to-peer search indexing
  • Federated AI model training
  • Distributed caching
  • Edge computing support

Technology Migration Path:

# Current (v1.x)
Architecture: Monolithic Python CLI
Storage: Local JSON files
Networking: Direct HTTP requests

# Planned (v2.x)
Architecture: Microservices + CLI
Storage: SQLite + Redis cache
Networking: Async HTTP/WebSockets

# Future (v3.x)
Architecture: Distributed P2P
Storage: Distributed database
Networking: Libp2p + WebRTC

🎯 Architecture Principles Summary

Core Design Principles:

  1. Separation of Concerns: Each component has a single responsibility
  2. Loose Coupling: Components communicate through interfaces
  3. High Cohesion: Related functionality grouped together
  4. Open/Closed: Open for extension, closed for modification
  5. Dependency Inversion: Depend on abstractions, not concretions
  6. Interface Segregation: Many specific interfaces vs one general
  7. Single Responsibility: Each class has one reason to change

Quality Attributes:

  • Performance: Caching, lazy loading, connection pooling
  • Scalability: Stateless components, horizontal scaling
  • Reliability: Error recovery, circuit breakers, retry logic
  • Security: Defense in depth, input validation, encryption
  • Maintainability: Clear interfaces, comprehensive tests, documentation
  • Extensibility: Plugin architecture, configuration system
  • Usability: Intuitive CLI, helpful errors, progressive disclosure

Trade-offs Made:

Decision Benefit Trade-off
Monolithic CLI Simple deployment Harder to scale
JSON storage Human readable Not optimized for queries
Synchronous I/O Simpler code Lower concurrency
Regex parsing No dependencies Less robust than HTML parsers

Last updated: 12/22/2025
Architecture version: 2.0

Key Takeaways:

  • NaviDuck follows a layered, modular architecture
  • Clear separation between UI, business logic, and data layers
  • Designed for extensibility and maintainability
  • Built with privacy and security as first-class concerns
  • Architecture supports future evolution to distributed systems

The architecture balances simplicity with sophistication, providing a solid foundation for both current features and future expansion.

Clone this wiki locally