Upload 94 files

app/api/sessions.py

@@ -4,7 +4,7 @@ from typing import Any, Dict
 
 from app.db.database import get_db
 from app.db import models, schemas
-from app.core.auth import get_current_active_user
+from app.core.dependencies import get_current_active_user
 
 router = APIRouter(
     prefix="/api/sessions",

app/api/staff_analytics.py

@@ -6,6 +6,7 @@ from ..core.dependencies import get_current_active_user, get_db
 from ..db.models import User, ActivityType
 from ..services.analytics import staff_analytics
 from ..services.staff_reports import staff_reports
+from ..services.performance_notifications import performance_notifications
 from ..db.schemas import (
     StaffActivityCreate,
     StaffActivityInDB,
@@ -25,13 +26,22 @@ async def record_staff_activity(
     if not current_user.branch_id:
         raise HTTPException(status_code=400, detail="User is not associated with a branch")
     
-    recorded_activity = await staff_analytics.record_activity(
+    recorded_activity, prev_metrics, new_metrics = await staff_analytics.record_activity(
         user_id=current_user.id,
         branch_id=current_user.branch_id,
         activity_type=activity.activity_type,
         details=activity.details,
         duration=activity.duration
     )
+
+    # Process notifications for the activity
+    await performance_notifications.process_activity_notifications(
+        user_id=current_user.id,
+        branch_id=current_user.branch_id,
+        activity=recorded_activity,
+        prev_metrics=prev_metrics,
+        new_metrics=new_metrics
+    )
     
     return {
         "status": "success",

app/main.py

@@ -8,7 +8,7 @@ from app.utils.logger import log_api_request
 from app.utils.tasks import run_periodic_tasks, sync_pos_metrics_task
 from app.services.websocket import connect, disconnect
 from app.realtime.subscriber import subscribe_order_events
-from app.routes.websocket import websocket_endpoint, manager, router as websocket_router
+from app.routes.websocket import websocket_endpoint, manager, router as websocket_router, staff_metrics_websocket
 import socketio
 import time
 import logging
@@ -65,16 +65,11 @@ async def health_check() -> Dict[str, str]:
 
 # WebSocket endpoint
 @app.websocket("/ws")
-async def websocket_endpoint(websocket: WebSocket):
-    await connect(websocket)
-    try:
-        while True:
-            data = await websocket.receive_text()
-    except:
-        await disconnect(websocket)
+async def websocket_handler(websocket: WebSocket):
+    await websocket_endpoint(websocket)
 
 @app.websocket("/ws/orders")
-async def orders_websocket_endpoint(websocket: WebSocket):
+async def orders_websocket_handler(websocket: WebSocket):
     await websocket_endpoint(websocket)
 
 # Request logging and rate limiting middleware

app/realtime/subscriber.py

@@ -0,0 +1,46 @@
+import asyncio
+from typing import Dict, Any
+import json
+from ..core.config import settings
+from ..utils.logger import logger
+import redis.asyncio as redis
+from ..routes.websocket import ConnectionManager
+
+async def subscribe_order_events(manager: ConnectionManager) -> None:
+    """Subscribe to Redis order events channel and broadcast updates to WebSocket clients"""
+    try:
+        r = redis.Redis(
+            host=settings.REDIS_HOST,
+            port=settings.REDIS_PORT,
+            decode_responses=True
+        )
+        
+        pubsub = r.pubsub()
+        await pubsub.subscribe("order_events")
+        
+        logger.info("Started Redis subscriber for order events")
+        
+        while True:
+            try:
+                message = await pubsub.get_message(ignore_subscribe_messages=True)
+                if message and message["type"] == "message":
+                    try:
+                        data = json.loads(message["data"])
+                        # Broadcast to all connected WebSocket clients
+                        await manager.broadcast(data)
+                        logger.debug(f"Broadcasted order event: {data}")
+                    except json.JSONDecodeError:
+                        logger.error(f"Invalid JSON in order event: {message['data']}")
+                    except Exception as e:
+                        logger.error(f"Error processing order event: {str(e)}")
+                
+                # Small delay to prevent tight loop
+                await asyncio.sleep(0.1)
+                
+            except Exception as e:
+                logger.error(f"Error in Redis subscription loop: {str(e)}")
+                await asyncio.sleep(1)  # Longer delay on error
+                
+    except Exception as e:
+        logger.error(f"Redis subscription error: {str(e)}")
+        raise  # Re-raise to allow startup to handle the error

app/routes/websocket.py

@@ -47,6 +47,14 @@ class ConnectionManager:
 
 manager = ConnectionManager()
 
+async def websocket_endpoint(websocket: WebSocket):
+    await manager.connect(websocket, "orders")
+    try:
+        while True:
+            data = await websocket.receive_text()
+    except WebSocketDisconnect:
+        manager.disconnect(websocket, "orders")
+
 @router.websocket("/ws/staff")
 async def staff_metrics_websocket(websocket: WebSocket):
     await manager.connect(websocket, "staff_metrics")

app/services/analytics.py

@@ -3,7 +3,6 @@ from sqlalchemy import func, and_, select, extract
 from ..db.database import db
 from ..utils.cache import cache
 from ..db.models import StaffActivity, PerformanceMetric, User, ActivityType
-from ..services.performance_notifications import performance_notifications
 from typing import Dict, List, Any, Optional
 import numpy as np
 from collections import defaultdict
@@ -180,22 +179,7 @@ class StaffAnalyticsService:
                 user_id, branch_id, activity
             )
             
-            # Check and send notifications based on new metrics
-            await performance_notifications.check_and_notify_performance(
-                user_id=user_id,
-                branch_id=branch_id,
-                metrics=new_metrics
-            )
-            
-            # Check for realtime alerts by comparing with previous metrics
-            if prev_metrics:
-                await performance_notifications.notify_realtime_alerts(
-                    branch_id=branch_id,
-                    current_metrics=new_metrics,
-                    previous_metrics=prev_metrics
-                )
-            
-            return activity
+            return activity, prev_metrics, new_metrics
 
     @staticmethod
     async def _get_current_metrics(user_id: int, branch_id: int) -> Optional[Dict[str, Any]]:

app/services/maintenance.py

@@ -11,9 +11,9 @@ from ..services.websocket import create_and_broadcast_notification
 from ..db.models import User, Order, Notification, Session
 from ..utils.cache import cache
 from sqlalchemy import text
-import aioredis
+import redis.asyncio as redis
 import asyncio
-from ..utils.retry import retry_with_backoff, circuit_breaker, CircuitBreaker
+from ..utils.retry import retry_with_backoff, circuit_breaker
 
 class MaintenanceService:
     def __init__(self):
@@ -21,10 +21,6 @@ class MaintenanceService:
         self._background_tasks = {}
         self.last_health_check = None
         self.last_maintenance = None
-        self.circuit_breaker = CircuitBreaker(
-            failure_threshold=5,
-            reset_timeout=60
-        )
 
     def get_timestamp(self) -> str:
         """Get current timestamp in ISO format"""
@@ -104,7 +100,7 @@ class MaintenanceService:
                 "error": str(e)
             }
 
-    @circuit_breaker(failure_threshold=5, reset_timeout=300.0)
+    @circuit_breaker(failure_threshold=5, reset_timeout=60)
     async def check_background_tasks(self) -> Dict[str, Any]:
         """Check status of background tasks"""
         active_tasks = []

app/services/performance_notifications.py

@@ -1,7 +1,6 @@
 from datetime import datetime
 from typing import Dict, Any, List, Optional
 from ..services.notifications import create_and_broadcast_notification
-from ..services.analytics import staff_analytics
 from ..db.models import ActivityType
 from ..utils.logger import logger
 
@@ -192,4 +191,26 @@ class PerformanceNotificationService:
         except Exception as e:
             logger.error(f"Error in realtime alert notifications: {str(e)}")
 
+    @staticmethod
+    async def process_activity_notifications(
+        user_id: int,
+        branch_id: int,
+        activity: Any,
+        prev_metrics: Optional[Dict[str, Any]],
+        new_metrics: Dict[str, Any]
+    ):
+        """Process notifications for a new staff activity"""
+        await PerformanceNotificationService.check_and_notify_performance(
+            user_id=user_id,
+            branch_id=branch_id,
+            metrics=new_metrics
+        )
+        
+        if prev_metrics:
+            await PerformanceNotificationService.notify_realtime_alerts(
+                branch_id=branch_id,
+                current_metrics=new_metrics,
+                previous_metrics=prev_metrics
+            )
+
 performance_notifications = PerformanceNotificationService()

app/services/pos_analytics.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Dict, Any, Optional, List
 from ..core.config import settings
 from ..utils.logger import logger, log_health_check
@@ -6,6 +6,7 @@ from ..services.analytics import staff_analytics
 from ..db.models import ActivityType
 from ..routes.websocket import broadcast_staff_update
 from ..utils.retry import with_retry, retry_with_backoff
+from ..services.performance_notifications import performance_notifications
 import httpx
 import asyncio
 import aiohttp
@@ -25,9 +26,7 @@ class POSAnalyticsService:
 
     @with_retry(max_retries=3, delay=1.0, exceptions=(httpx.HTTPError, asyncio.TimeoutError))
     async def sync_pos_activity(self, pos_data: Dict[str, Any]) -> bool:
-        """
-        Sync staff activity data from POS system with retry mechanism
-        """
+        """Sync staff activity data from POS system with retry mechanism"""
         try:
             # Map POS activity types to our ActivityType enum
             activity_mapping = {
@@ -46,7 +45,7 @@ class POSAnalyticsService:
                 return False
 
             # Record the activity in our system
-            await staff_analytics.record_activity(
+            activity, prev_metrics, new_metrics = await staff_analytics.record_activity(
                 user_id=pos_data["user_id"],
                 branch_id=pos_data["branch_id"],
                 activity_type=activity_type,
@@ -54,6 +53,15 @@ class POSAnalyticsService:
                 duration=pos_data.get("duration")
             )
 
+            # Process notifications for the activity
+            await performance_notifications.process_activity_notifications(
+                user_id=pos_data["user_id"],
+                branch_id=pos_data["branch_id"],
+                activity=activity,
+                prev_metrics=prev_metrics,
+                new_metrics=new_metrics
+            )
+
             # Get updated metrics for the branch and broadcast
             metrics = await staff_analytics.get_staff_performance(
                 branch_id=pos_data["branch_id"],
@@ -93,61 +101,66 @@ class POSAnalyticsService:
             logger.error(f"Error fetching POS metrics: {str(e)}")
             return None
 
-    @with_retry(max_retries=3, delay=2.0, exceptions=(httpx.HTTPError, asyncio.TimeoutError))
+    @retry_with_backoff(max_retries=3)
     async def sync_all_metrics(self, branch_id: Optional[int] = None) -> bool:
-        """
-        Sync all staff metrics from POS system with retry mechanism
-        """
+        """Sync all metrics from POS system"""
         try:
-            today = datetime.utcnow()
-            params = {"date": today.date().isoformat()}
-            if branch_id:
-                params["branch_id"] = branch_id
+            # Reset failed operations list
+            self._failed_operations = []
+            
+            # Sync various metrics
+            tasks = [
+                self._sync_sales_metrics(),
+                self._sync_inventory_metrics(),
+                self._sync_staff_metrics(),
+                self._sync_customer_metrics()
+            ]
+            
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            
+            # Check for failures
+            for result in results:
+                if isinstance(result, Exception):
+                    self._failed_operations.append(str(result))
+                    logger.error(f"Metric sync failed: {str(result)}")
+            
+            self._last_sync_time = datetime.utcnow()
+            self._last_error = None if not self._failed_operations else self._failed_operations[-1]
 
-            async with httpx.AsyncClient() as client:
-                response = await client.get(
-                    f"{self.pos_api_url}/api/v1/staff/metrics/daily",
-                    headers=self._headers,
-                    params=params,
-                    timeout=30.0
-                )
-                
-                response.raise_for_status()
-                metrics_data = response.json()
-                sync_tasks = []
-                processed_branches = set()
+            # Process metrics and sync staff activities
+            sync_tasks = []
+            processed_branches = set()
 
-                for metric in metrics_data:
-                    activities = self._convert_metrics_to_activities(metric)
+            for metric in (r for r in results if not isinstance(r, Exception)):
+                activities = self._convert_metrics_to_activities(metric)
+                if "branch_id" in metric:
                     processed_branches.add(metric["branch_id"])
-                    
-                    for activity in activities:
-                        sync_tasks.append(self.sync_pos_activity(activity))
-
-                # Run all sync tasks concurrently with individual retries
-                results = await asyncio.gather(*sync_tasks, return_exceptions=True)
                 
-                # Broadcast updates for each affected branch
-                for branch_id in processed_branches:
-                    metrics = await staff_analytics.get_staff_performance(branch_id=branch_id)
-                    await broadcast_staff_update(metrics)
-
-                # Check for any failures
-                success = all(
-                    result is True if not isinstance(result, Exception) else False
-                    for result in results
-                )
+                for activity in activities:
+                    sync_tasks.append(self.sync_pos_activity(activity))
 
-                if not success:
-                    logger.warning("Some metrics failed to sync")
+            # Run all sync tasks concurrently with individual retries
+            activity_results = await asyncio.gather(*sync_tasks, return_exceptions=True)
+            
+            # Broadcast updates for each affected branch
+            for branch_id in processed_branches:
+                metrics = await staff_analytics.get_staff_performance(branch_id=branch_id)
+                await broadcast_staff_update(metrics)
+
+            # Check overall success
+            success = len(self._failed_operations) == 0 and all(
+                result is True if not isinstance(result, Exception) else False
+                for result in activity_results
+            )
 
-                return success
+            if not success:
+                logger.warning("Some metrics failed to sync")
+
+            return success
 
-        except httpx.HTTPError as e:
-            logger.error(f"HTTP error in sync_all_metrics: {str(e)}")
-            raise
         except Exception as e:
-            logger.error(f"Error in sync_all_metrics: {str(e)}")
+            self._last_error = str(e)
+            logger.error(f"Failed to sync metrics: {str(e)}")
             return False
 
     def _convert_metrics_to_activities(self, metric: Dict[str, Any]) -> List[Dict[str, Any]]:
@@ -258,39 +271,6 @@ class POSAnalyticsService:
                 "details": {"error": str(e)}
             }
 
-    @retry_with_backoff(max_retries=3)
-    async def sync_all_metrics(self) -> bool:
-        """Sync all metrics from POS system"""
-        try:
-            # Reset failed operations list
-            self._failed_operations = []
-            
-            # Sync various metrics
-            tasks = [
-                self._sync_sales_metrics(),
-                self._sync_inventory_metrics(),
-                self._sync_staff_metrics(),
-                self._sync_customer_metrics()
-            ]
-            
-            results = await asyncio.gather(*tasks, return_exceptions=True)
-            
-            # Check for failures
-            for result in results:
-                if isinstance(result, Exception):
-                    self._failed_operations.append(str(result))
-                    logger.error(f"Metric sync failed: {str(result)}")
-            
-            self._last_sync_time = datetime.utcnow()
-            self._last_error = None if not self._failed_operations else self._failed_operations[-1]
-            
-            return len(self._failed_operations) == 0
-
-        except Exception as e:
-            self._last_error = str(e)
-            logger.error(f"Failed to sync metrics: {str(e)}")
-            return False
-
     @retry_with_backoff(max_retries=2)
     async def _sync_sales_metrics(self) -> Dict[str, Any]:
         """Sync sales metrics from POS"""

app/utils/cache.py

@@ -1,106 +1,32 @@
-import redis
+import redis.asyncio as redis
 import json
 import inspect
 import functools
 from ..core.config import settings
 from typing import Any, Optional, Callable, TypeVar
 from ..utils.logger import logger
-from redis.asyncio import Redis
-import asyncio
 
 T = TypeVar('T')
 
-def cached(ttl_seconds: int):
-    """
-    Cache decorator that stores function results in Redis.
-    Works with both sync and async functions.
-    
-    Args:
-        ttl_seconds: Time to live in seconds for cached results
-        
-    Example:
-        @cached(300)  # Cache for 5 minutes
-        async def get_user(user_id: int):
-            return await db.fetch_user(user_id)
-    """
-    def decorator(func: Callable[..., T]) -> Callable[..., T]:
-        is_async = inspect.iscoroutinefunction(func)
-        
-        def get_cache_key(*args, **kwargs) -> str:
-            """Generate cache key from function name and arguments"""
-            # Sort kwargs to ensure consistent key generation
-            sorted_kwargs = sorted(kwargs.items())
-            args_str = ":".join(str(arg) for arg in args)
-            kwargs_str = ":".join(f"{k}={v}" for k, v in sorted_kwargs)
-            return f"cache:{func.__module__}:{func.__name__}:{args_str}:{kwargs_str}"
-
-        if is_async:
-            @functools.wraps(func)
-            async def async_wrapper(*args, **kwargs) -> T:
-                cache_key = get_cache_key(*args, **kwargs)
-                
-                # Try to get from cache
-                cached_value = await cache.get_cache(cache_key)
-                if cached_value is not None:
-                    return cached_value
-                
-                # Call function and cache result
-                result = await func(*args, **kwargs)
-                await cache.set_cache(cache_key, result, ttl_seconds)
-                return result
-            return async_wrapper
-        else:
-            @functools.wraps(func)
-            def sync_wrapper(*args, **kwargs) -> T:
-                cache_key = get_cache_key(*args, **kwargs)
-                
-                # Try to get from cache
-                try:
-                    cached_value = cache.redis_client.get(cache_key)
-                    if cached_value:
-                        return json.loads(cached_value)
-                except:
-                    if cache.is_connected:
-                        logger.error("Redis error in sync cache access")
-                    return cache.fallback_cache.get(cache_key)
-                
-                # Call function and cache result
-                result = func(*args, **kwargs)
-                try:
-                    if cache.is_connected:
-                        cache.redis_client.setex(
-                            cache_key,
-                            ttl_seconds,
-                            json.dumps(result)
-                        )
-                    else:
-                        cache.fallback_cache[cache_key] = result
-                except Exception as e:
-                    logger.error(f"Cache set error in sync wrapper: {str(e)}")
-                return result
-            return sync_wrapper
-            
-    return decorator
-
 class RedisCache:
     _instance = None
 
     def __new__(cls):
         if cls._instance is None:
             cls._instance = super(RedisCache, cls).__new__(cls)
-            asyncio.create_task(cls._instance.initialize())
+            cls._instance.is_connected = False
+            cls._instance.fallback_cache = {}
+            cls._instance.redis = redis.Redis(
+                host=settings.REDIS_HOST,
+                port=settings.REDIS_PORT,
+                decode_responses=True
+            )
         return cls._instance
 
     async def initialize(self):
         """Initialize Redis connection with fallback to dummy cache"""
         try:
-            self.redis_client = Redis(
-                host=settings.REDIS_HOST,
-                port=settings.REDIS_PORT,
-                decode_responses=True,
-                socket_connect_timeout=1
-            )
-            await self.redis_client.ping()
+            await self.redis.ping()
             self.is_connected = True
             logger.info("Redis cache initialized successfully")
         except Exception as e:
@@ -110,74 +36,101 @@ class RedisCache:
 
     async def set_cache(self, key: str, value: Any, expire: int = 3600):
         """Set a cache entry with optional expiration time (default 1 hour)"""
-        if not self.is_connected:
-            self.fallback_cache[key] = value
-            return True
-
         try:
-            serialized_value = json.dumps(value)
-            await self.redis_client.set(key, serialized_value, ex=expire)
-            return True
+            if self.is_connected:
+                await self.redis.set(key, json.dumps(value), ex=expire)
+            else:
+                self.fallback_cache[key] = value
         except Exception as e:
-            logger.error(f"Error setting cache: {str(e)}")
-            return False
+            logger.error(f"Cache set error: {str(e)}")
+            self.fallback_cache[key] = value
 
     async def get_cache(self, key: str) -> Optional[Any]:
         """Get a cached value by key"""
-        if not self.is_connected:
-            return self.fallback_cache.get(key)
-
         try:
-            value = await self.redis_client.get(key)
-            if value:
-                return json.loads(value)
+            if self.is_connected:
+                value = await self.redis.get(key)
+                return json.loads(value) if value else None
+            return self.fallback_cache.get(key)
         except Exception as e:
-            logger.error(f"Error getting cache: {str(e)}")
-        return None
+            logger.error(f"Cache get error: {str(e)}")
+            return self.fallback_cache.get(key)
 
     async def delete_cache(self, key: str) -> bool:
         """Delete a cache entry by key"""
-        if not self.is_connected:
-            self.fallback_cache.pop(key, None)
-            return True
-
         try:
-            await self.redis_client.delete(key)
+            if self.is_connected:
+                return bool(await self.redis.delete(key))
+            self.fallback_cache.pop(key, None)
             return True
         except Exception as e:
-            logger.error(f"Error deleting cache: {str(e)}")
+            logger.error(f"Cache delete error: {str(e)}")
             return False
 
     async def clear_cache_pattern(self, pattern: str) -> bool:
         """Clear all cache entries matching a pattern"""
-        if not self.is_connected:
-            # Basic pattern matching for in-memory cache
-            keys_to_delete = [k for k in self.fallback_cache.keys() if pattern in k]
-            for key in keys_to_delete:
-                del self.fallback_cache[key]
-            return True
-
         try:
-            cursor = 0
-            while True:
-                cursor, keys = await self.redis_client.scan(cursor, match=pattern)
+            if self.is_connected:
+                keys = await self.redis.keys(pattern)
                 if keys:
-                    await self.redis_client.delete(*keys)
-                if cursor == 0:
-                    break
+                    return bool(await self.redis.delete(*keys))
             return True
         except Exception as e:
-            logger.error(f"Error clearing cache pattern: {str(e)}")
+            logger.error(f"Cache clear pattern error: {str(e)}")
             return False
 
     async def check_connection(self) -> bool:
         """Check if Redis connection is alive"""
         try:
-            await self.redis_client.ping()
-            self.is_connected = True
-            return True
-        except:
-            self.is_connected = False
+            if self.is_connected:
+                await self.redis.ping()
+                return True
             return False
+        except Exception:
+            return False
+
+def cached(ttl_seconds: int):
+    """Decorator to cache function results"""
+    def decorator(func: Callable[..., T]) -> Callable[..., T]:
+        if inspect.iscoroutinefunction(func):
+            @functools.wraps(func)
+            async def async_wrapper(*args, **kwargs) -> T:
+                cache_key = f"{func.__name__}:{args}:{kwargs}"
+                cached_value = await cache.get_cache(cache_key)
+                if cached_value is not None:
+                    return cached_value
+                
+                result = await func(*args, **kwargs)
+                await cache.set_cache(cache_key, result, expire=ttl_seconds)
+                return result
+            return async_wrapper
+        else:
+            @functools.wraps(func)
+            def sync_wrapper(*args, **kwargs) -> T:
+                cache_key = f"{func.__name__}:{args}:{kwargs}"
+                try:
+                    cached_value = cache.redis.get(cache_key)
+                    if cached_value:
+                        return json.loads(cached_value)
+                except:
+                    if cache.is_connected:
+                        logger.error("Redis error in sync cache access")
+                    return cache.fallback_cache.get(cache_key)
+                
+                result = func(*args, **kwargs)
+                try:
+                    if cache.is_connected:
+                        cache.redis.setex(
+                            cache_key,
+                            ttl_seconds,
+                            json.dumps(result)
+                        )
+                    else:
+                        cache.fallback_cache[cache_key] = result
+                except Exception as e:
+                    logger.error(f"Cache set error in sync wrapper: {str(e)}")
+                return result
+            return sync_wrapper
+    return decorator
 
 cache = RedisCache()

app/utils/tasks.py

@@ -1,9 +1,9 @@
+import asyncio
 from datetime import datetime, timedelta
 from ..db.database import db
-from ..services.maintenance import maintenance
+from ..services.maintenance import maintenance_service
 from ..utils.logger import logger
 from ..services.websocket import create_and_broadcast_notification
-import asyncio
 from sqlalchemy import select, delete
 from ..db.models import Event, User, Notification
 from ..services.pos_analytics import pos_analytics
@@ -73,23 +73,23 @@ async def perform_daily_maintenance():
     """Perform daily system maintenance tasks"""
     try:
         # Clean up expired sessions
-        deleted_sessions = await maintenance.cleanup_expired_sessions()
+        deleted_sessions = await maintenance_service.cleanup_expired_sessions()
         logger.info(f"Cleaned up {deleted_sessions} expired sessions")
 
         # Archive old data
-        archived = await maintenance.archive_old_data()
+        archived = await maintenance_service.archive_old_data()
         if archived:
             logger.info(f"Archived data: {archived}")
 
         # Check system health
-        health_data = await maintenance.check_system_health()
+        health_data = await maintenance_service.check_system_health()
         if "error" not in health_data:
             logger.info("System health check completed successfully")
         else:
             logger.error(f"System health check error: {health_data['error']}")
 
         # Monitor system resources
-        resources = await maintenance.monitor_system_resources()
+        resources = await maintenance_service.monitor_system_resources()
         if "error" not in resources:
             logger.info("System resource monitoring completed successfully")
         else:
@@ -102,15 +102,15 @@ async def perform_weekly_maintenance():
     """Perform weekly system maintenance tasks"""
     try:
         # Perform database maintenance
-        await maintenance.perform_database_maintenance()
+        await maintenance_service.perform_database_maintenance()
         logger.info("Database maintenance completed successfully")
 
         # Rotate log files
-        await maintenance.rotate_log_files()
+        await maintenance_service.rotate_log_files()
         logger.info("Log rotation completed successfully")
 
         # Manage storage quotas
-        quota_results = await maintenance.manage_storage_quotas()
+        quota_results = await maintenance_service.manage_storage_quotas()
         if quota_results.get("warnings"):
             for warning in quota_results["warnings"]:
                 logger.warning(warning)