Upload 82 files

app/core/config.py

@@ -1,34 +1,34 @@
-from pydantic_settings import BaseSettings
-from typing import Optional
-
-class Settings(BaseSettings):
-    API_V1_STR: str = "/api/v1"
-    PROJECT_NAME: str = "Admin Dashboard"
-    VERSION: str = "1.0.0"
-    
-    # PostgreSQL Database settings
-    DATABASE_URL: str = "postgresql+asyncpg://postgres:Lovyelias5584.@db.mqyrkmsdgugdhxiucukb.supabase.co:5432/postgres"
-    
-    # JWT Settings
-    SECRET_KEY: str = "your-secret-key-here"  # Change in production
-    ALGORITHM: str = "HS256"
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
-    
-    # Redis settings
-    REDIS_HOST: str = "localhost"
-    REDIS_PORT: int = 6379
-    
-    # Email settings
-    MAIL_USERNAME: str = "yungdml31@gmail.com"
-    MAIL_PASSWORD: str = ""
-    MAIL_FROM: str = "admin@angelo.com"
-    MAIL_PORT: int = 587
-    MAIL_SERVER: str = "smtp.gmail.com"
-    
-    # Frontend URL for email links
-    FRONTEND_URL: str = "http://localhost:3000"
-    
-    class Config:
-        case_sensitive = True
-
+from pydantic_settings import BaseSettings
+from typing import Optional
+
+class Settings(BaseSettings):
+    API_V1_STR: str = "/api/v1"
+    PROJECT_NAME: str = "Admin Dashboard"
+    VERSION: str = "1.0.0"
+    
+    # PostgreSQL Database settings
+    DATABASE_URL: str = "postgresql+asyncpg://postgres:Lovyelias5584.@db.mqyrkmsdgugdhxiucukb.supabase.co:5432/postgres"
+    
+    # JWT Settings
+    SECRET_KEY: str = "your-secret-key-here"  # Change in production
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+    
+    # Redis settings
+    REDIS_HOST: str = "localhost"
+    REDIS_PORT: int = 6379
+    
+    # Email settings
+    MAIL_USERNAME: str = "yungdml31@gmail.com"
+    MAIL_PASSWORD: str = ""
+    MAIL_FROM: str = "admin@angelo.com"
+    MAIL_PORT: int = 587
+    MAIL_SERVER: str = "smtp.gmail.com"
+    
+    # Frontend URL for email links
+    FRONTEND_URL: str = "http://localhost:3000"
+    
+    class Config:
+        case_sensitive = True
+
 settings = Settings()

app/db/database.py

@@ -1,32 +1,32 @@
-from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
-from sqlalchemy.orm import declarative_base
-from ..core.config import settings
-
-# Create async engine for FastAPI
-async_engine = create_async_engine(
-    settings.DATABASE_URL,
-    echo=True,
-    future=True,
-    pool_pre_ping=True
-)
-
-# Create async session factory
-AsyncSessionLocal = async_sessionmaker(
-    bind=async_engine,
-    class_=AsyncSession,
-    expire_on_commit=False
-)
-
-# Create declarative base for models
-Base = declarative_base()
-
-# Database dependency for FastAPI routes
-async def get_db():
-    async with AsyncSessionLocal() as session:
-        try:
-            yield session
-        finally:
-            await session.close()
-
-# Add alias for external imports
-db = AsyncSessionLocal
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+from sqlalchemy.orm import declarative_base
+from ..core.config import settings
+
+# Create async engine for FastAPI
+async_engine = create_async_engine(
+    settings.DATABASE_URL,
+    echo=True,
+    future=True,
+    pool_pre_ping=True
+)
+
+# Create async session factory
+AsyncSessionLocal = async_sessionmaker(
+    bind=async_engine,
+    class_=AsyncSession,
+    expire_on_commit=False
+)
+
+# Create declarative base for models
+Base = declarative_base()
+
+# Database dependency for FastAPI routes
+async def get_db():
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+        finally:
+            await session.close()
+
+# Add alias for external imports
+db = AsyncSessionLocal

app/main.py

@@ -1,108 +1,95 @@
-from fastapi import FastAPI, Request, WebSocket
-from fastapi.middleware.cors import CORSMiddleware
-from .core.config import settings
-from .db.database import async_engine as engine, Base
-from .api import auth, products, orders, users, analytics, files, notifications, calendar, scheduler, maintenance
-from .utils.rate_limiter import rate_limiter
-from .utils.logger import log_api_request
-from .utils.tasks import run_periodic_tasks
-import time
-import logging
-import asyncio
-from typing import List
-
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-app = FastAPI(title=settings.PROJECT_NAME, version=settings.VERSION)
-
-# Store active WebSocket connections and background tasks
-active_connections: List[WebSocket] = []
-background_tasks = set()
-
-# Configure CORS
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],  # Configure appropriately for production
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# WebSocket connection manager
-@app.websocket("/ws")
-async def websocket_endpoint(websocket: WebSocket):
-    await websocket.accept()
-    active_connections.append(websocket)
-    try:
-        while True:
-            data = await websocket.receive_text()
-    except:
-        active_connections.remove(websocket)
-
-# Notification broadcaster
-async def broadcast_notification(message: dict):
-    for connection in active_connections:
-        try:
-            await connection.send_json(message)
-        except:
-            active_connections.remove(connection)
-
-# Request logging and rate limiting middleware
-@app.middleware("http")
-async def middleware(request: Request, call_next):
-    await rate_limiter.check_rate_limit(request)
-    start_time = time.time()
-    response = await call_next(request)
-    end_time = time.time()
-    duration = end_time - start_time
-    log_api_request(
-        method=request.method,
-        path=request.url.path,
-        status_code=response.status_code,
-        duration=duration
-    )
-    return response
-
-# Application startup and shutdown events
-@app.on_event("startup")
-async def startup_event():
-    # Create all database tables
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
-    
-    # Start background tasks
-    task = asyncio.create_task(run_periodic_tasks())
-    background_tasks.add(task)
-    task.add_done_callback(background_tasks.discard)
-
-@app.on_event("shutdown")
-async def shutdown_event():
-    # Cancel background tasks
-    for task in background_tasks:
-        task.cancel()
-    
-    # Close WebSocket connections
-    for connection in active_connections:
-        await connection.close()
-
-# Include routers
-app.include_router(auth.router, prefix=f"{settings.API_V1_STR}/auth", tags=["auth"])
-app.include_router(users.router, prefix=f"{settings.API_V1_STR}/users", tags=["users"])
-app.include_router(products.router, prefix=f"{settings.API_V1_STR}/products", tags=["products"])
-app.include_router(orders.router, prefix=f"{settings.API_V1_STR}/orders", tags=["orders"])
-app.include_router(analytics.router, prefix=f"{settings.API_V1_STR}/analytics", tags=["analytics"])
-app.include_router(files.router, prefix=f"{settings.API_V1_STR}/files", tags=["files"])
-app.include_router(notifications.router, prefix=f"{settings.API_V1_STR}/notifications", tags=["notifications"])
-app.include_router(calendar.router, prefix=f"{settings.API_V1_STR}/calendar", tags=["calendar"])
-app.include_router(scheduler.router, prefix=f"{settings.API_V1_STR}/scheduler", tags=["scheduler"])
-app.include_router(maintenance.router, prefix=f"{settings.API_V1_STR}/maintenance", tags=["maintenance"])
-
-@app.get("/")
-async def root():
-    return {
-        "message": f"Welcome to {settings.PROJECT_NAME} v{settings.VERSION}",
-        "docs_url": "/docs",
-        "openapi_url": "/openapi.json"
+from fastapi import FastAPI, Request, WebSocket
+from fastapi.middleware.cors import CORSMiddleware
+from .core.config import settings
+from .db.database import async_engine as engine, Base
+from .api import auth, products, orders, users, analytics, files, notifications, calendar, scheduler, maintenance
+from .utils.rate_limiter import rate_limiter
+from .utils.logger import log_api_request
+from .utils.tasks import run_periodic_tasks
+from .services.websocket import connect, disconnect
+import time
+import logging
+import asyncio
+from typing import List
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(title=settings.PROJECT_NAME, version=settings.VERSION)
+
+# Store background tasks
+background_tasks = set()
+
+# Configure CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Configure appropriately for production
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 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)
+
+# Request logging and rate limiting middleware
+@app.middleware("http")
+async def middleware(request: Request, call_next):
+    await rate_limiter.check_rate_limit(request)
+    start_time = time.time()
+    response = await call_next(request)
+    end_time = time.time()
+    duration = end_time - start_time
+    log_api_request(
+        method=request.method,
+        path=request.url.path,
+        status_code=response.status_code,
+        duration=duration
+    )
+    return response
+
+# Application startup and shutdown events
+@app.on_event("startup")
+async def startup_event():
+    # Create all database tables
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    
+    # Start background tasks
+    task = asyncio.create_task(run_periodic_tasks())
+    background_tasks.add(task)
+    task.add_done_callback(background_tasks.discard)
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    # Cancel background tasks
+    for task in background_tasks:
+        task.cancel()
+
+# Include routers
+app.include_router(auth.router, prefix=f"{settings.API_V1_STR}/auth", tags=["auth"])
+app.include_router(users.router, prefix=f"{settings.API_V1_STR}/users", tags=["users"])
+app.include_router(products.router, prefix=f"{settings.API_V1_STR}/products", tags=["products"])
+app.include_router(orders.router, prefix=f"{settings.API_V1_STR}/orders", tags=["orders"])
+app.include_router(analytics.router, prefix=f"{settings.API_V1_STR}/analytics", tags=["analytics"])
+app.include_router(files.router, prefix=f"{settings.API_V1_STR}/files", tags=["files"])
+app.include_router(notifications.router, prefix=f"{settings.API_V1_STR}/notifications", tags=["notifications"])
+app.include_router(calendar.router, prefix=f"{settings.API_V1_STR}/calendar", tags=["calendar"])
+app.include_router(scheduler.router, prefix=f"{settings.API_V1_STR}/scheduler", tags=["scheduler"])
+app.include_router(maintenance.router, prefix=f"{settings.API_V1_STR}/maintenance", tags=["maintenance"])
+
+@app.get("/")
+async def root():
+    return {
+        "message": f"Welcome to {settings.PROJECT_NAME} v{settings.VERSION}",
+        "docs_url": "/docs",
+        "openapi_url": "/openapi.json"
     }

app/services/maintenance.py

@@ -1,286 +1,202 @@
 import os
-import psutil
+import shutil
 from datetime import datetime, timedelta
-from typing import Dict, Any, List
-
+from typing import Dict, Any, Optional
+from sqlalchemy import select, delete, text
 from ..db.database import db
 from ..utils.logger import logger
-from ..utils.cache import cache
-from ..services.notifications import notifications
-from .backup import backup
-import gzip
-import shutil
+from ..core.config import settings
+from ..services.websocket import create_and_broadcast_notification
 
 class MaintenanceService:
     async def cleanup_expired_sessions(self) -> int:
-        """Clean up expired sessions from the database"""
-        try:
-            cutoff_date = datetime.utcnow() - timedelta(days=7)
-            result = await db.db["sessions"].delete_many({
-                "last_activity": {"$lt": cutoff_date}
-            })
-            logger.info(f"Cleaned up {result.deleted_count} expired sessions")
-            return result.deleted_count
-        except Exception as e:
-            logger.error(f"Error cleaning up sessions: {str(e)}")
-            return 0
+        """Clean up expired sessions"""
+        cutoff = datetime.utcnow() - timedelta(days=7)
+        result = await db.db["sessions"].delete_many({
+            "last_activity": {"$lt": cutoff}
+        })
+        return result.deleted_count
 
     async def archive_old_data(self) -> Dict[str, int]:
-        """Archive old data to maintain database performance"""
+        """Archive old data"""
         try:
-            archive_date = datetime.utcnow() - timedelta(days=365)  # Archive data older than 1 year
-            archives = {}
+            cutoff = datetime.utcnow() - timedelta(days=365)
+            archived = {}
 
             # Archive old orders
-            old_orders = await db.db["orders"].find({
-                "created_at": {"$lt": archive_date},
-                "status": {"$in": ["delivered", "cancelled"]}
-            }).to_list(None)
-
-            if old_orders:
-                await db.db["archived_orders"].insert_many(old_orders)
-                result = await db.db["orders"].delete_many({
-                    "_id": {"$in": [order["_id"] for order in old_orders]}
-                })
-                archives["orders"] = len(old_orders)
+            result = await db.db["orders"].update_many(
+                {
+                    "created_at": {"$lt": cutoff},
+                    "status": {"$in": ["completed", "cancelled"]}
+                },
+                {"$set": {"archived": True}}
+            )
+            archived["orders"] = result.modified_count
 
             # Archive old notifications
-            old_notifications = await db.db["notifications"].find({
-                "created_at": {"$lt": archive_date},
-                "read": True
-            }).to_list(None)
-
-            if old_notifications:
-                await db.db["archived_notifications"].insert_many(old_notifications)
-                result = await db.db["notifications"].delete_many({
-                    "_id": {"$in": [notif["_id"] for notif in old_notifications]}
-                })
-                archives["notifications"] = len(old_notifications)
-
-            # Archive old metrics (keep last 90 days)
-            cutoff_date = datetime.utcnow() - timedelta(days=90)
-            result = await db.db.system_metrics.delete_many({
-                "timestamp": {"$lt": cutoff_date}
-            })
-            logger.info(f"Archived {result.deleted_count} old metric records")
-            archives["metrics"] = result.deleted_count
-
-            # Archive old audit logs (keep last 180 days)
-            audit_cutoff = datetime.utcnow() - timedelta(days=180)
-            audit_result = await db.db.audit_logs.delete_many({
-                "timestamp": {"$lt": audit_cutoff}
-            })
-            logger.info(f"Archived {audit_result.deleted_count} old audit logs")
-            archives["audit_logs"] = audit_result.deleted_count
-
-            return archives
+            result = await db.db["notifications"].update_many(
+                {
+                    "created_at": {"$lt": cutoff},
+                    "read": True
+                },
+                {"$set": {"archived": True}}
+            )
+            archived["notifications"] = result.modified_count
+
+            return archived
         except Exception as e:
             logger.error(f"Error archiving old data: {str(e)}")
-            return {}
+            return None
 
     async def check_system_health(self) -> Dict[str, Any]:
-        """Check various system health metrics"""
+        """Check system health"""
         try:
-            health_data = {
-                "timestamp": datetime.utcnow(),
-                "database": {},
-                "cache": {},
-                "storage": {}
-            }
+            # Check database connection
+            await db.db.command("ping")
 
-            # Check database stats
-            db_stats = await db.db.command("dbStats")
-            health_data["database"] = {
-                "size": db_stats["dataSize"],
-                "collections": db_stats["collections"],
-                "indexes": db_stats["indexes"]
-            }
+            # Get database stats
+            stats = await db.db.command("dbStats")
+            disk_size = stats.get("dataSize", 0) / (1024 * 1024)  # Convert to MB
 
-            # Check Redis cache
-            try:
-                cache_info = await cache.redis_client.info()
-                health_data["cache"] = {
+            # Check disk space
+            total_space = shutil.disk_usage("/").total / (1024 * 1024 * 1024)  # GB
+            free_space = shutil.disk_usage("/").free / (1024 * 1024 * 1024)  # GB
+            
+            health_data = {
+                "status": "healthy",
+                "database": {
                     "connected": True,
-                    "used_memory": cache_info["used_memory"],
-                    "connected_clients": cache_info["connected_clients"]
-                }
-            except:
-                health_data["cache"] = {"connected": False}
-
-            # Check storage metrics
-            storage_stats = {
-                "uploads_size": await self._get_directory_size("uploads"),
-                "logs_size": await self._get_directory_size("logs")
+                    "size_mb": disk_size
+                },
+                "disk": {
+                    "total_gb": total_space,
+                    "free_gb": free_space,
+                    "usage_percent": ((total_space - free_space) / total_space) * 100
+                },
+                "timestamp": datetime.utcnow()
             }
-            health_data["storage"] = storage_stats
 
-            return health_data
-        except Exception as e:
-            logger.error(f"Error checking system health: {str(e)}")
-            return {"error": str(e)}
+            # Send alert if disk space is low
+            if free_space < 5:  # Less than 5GB free
+                await create_and_broadcast_notification(
+                    user_id="admin",
+                    title="Low Disk Space Alert",
+                    message=f"Server is running low on disk space. Only {free_space:.2f}GB remaining.",
+                    notification_type="system_alert",
+                    data={"free_space_gb": free_space}
+                )
 
-    async def perform_database_maintenance(self):
-        """Perform routine database maintenance tasks"""
-        try:
-            # Cleanup expired sessions
-            await self.cleanup_expired_sessions()
-            
-            # Run database vacuum and analyze
-            await db.db.command('analyze')
-            logger.info("Database maintenance completed successfully")
+            return health_data
         except Exception as e:
-            logger.error(f"Database maintenance failed: {str(e)}")
-            raise
+            logger.error(f"Health check error: {str(e)}")
+            return {"status": "unhealthy", "error": str(e)}
 
     async def monitor_system_resources(self) -> Dict[str, Any]:
-        """Monitor system resources and return metrics"""
+        """Monitor system resources"""
         try:
-            cpu_percent = psutil.cpu_percent()
-            memory = psutil.virtual_memory()
-            disk = psutil.disk_usage('/')
+            # Monitor database connections
+            server_status = await db.db.command("serverStatus")
+            current_connections = server_status.get("connections", {}).get("current", 0)
             
-            metrics = {
-                "cpu_usage": cpu_percent,
-                "memory_usage": memory.percent,
-                "disk_usage": disk.percent,
+            resources = {
+                "database": {
+                    "connections": current_connections,
+                },
                 "timestamp": datetime.utcnow()
             }
-            
-            await db.db.system_metrics.insert_one(metrics)
-            
-            # Alert if resources are critically low
-            if any([cpu_percent > 90, memory.percent > 90, disk.percent > 90]):
-                logger.warning("System resources critically low", extra=metrics)
-                
-            return metrics
-        except Exception as e:
-            logger.error(f"Resource monitoring failed: {str(e)}")
-            raise
 
-    async def perform_scheduled_backup(self):
-        """Perform scheduled system backup"""
-        try:
-            result = await backup.create_backup(include_files=True)
-            logger.info(f"Scheduled backup completed successfully: {result['id']}")
-            
-            # Cleanup old backups (keep last 7 days)
-            await self.cleanup_old_backups(days_to_keep=7)
+            # Alert if too many connections
+            if current_connections > settings.MAX_DB_CONNECTIONS * 0.9:
+                await create_and_broadcast_notification(
+                    user_id="admin",
+                    title="High Database Connections",
+                    message=f"Database has {current_connections} active connections",
+                    notification_type="system_alert",
+                    data={"connections": current_connections}
+                )
+
+            return resources
         except Exception as e:
-            logger.error(f"Scheduled backup failed: {str(e)}")
-            raise
+            logger.error(f"Resource monitoring error: {str(e)}")
+            return {"error": str(e)}
 
-    async def cleanup_old_backups(self, days_to_keep: int = 7):
-        """Clean up backups older than specified days"""
+    async def perform_database_maintenance(self) -> Dict[str, Any]:
+        """Perform database maintenance tasks"""
         try:
-            cutoff_date = datetime.utcnow() - timedelta(days=days_to_keep)
-            old_backups = await db.db.backup_history.find({
-                "created_at": {"$lt": cutoff_date}
-            }).to_list(None)
+            # Run VACUUM ANALYZE
+            await db.db.command("vacuum")
+            
+            # Clean up temporary collections
+            temp_collections = [
+                "temp_imports",
+                "temp_exports",
+                "temp_reports"
+            ]
+            for collection in temp_collections:
+                await db.db[collection].drop()
             
-            for old_backup in old_backups:
-                await backup.delete_backup(str(old_backup["_id"]))
-                
-            logger.info(f"Cleaned up {len(old_backups)} old backups")
+            return {"status": "success"}
         except Exception as e:
-            logger.error(f"Backup cleanup failed: {str(e)}")
-            raise
+            logger.error(f"Database maintenance error: {str(e)}")
+            return {"error": str(e)}
 
-    async def rotate_log_files(self):
-        """Rotate and archive log files"""
+    async def rotate_log_files(self) -> None:
+        """Rotate log files"""
+        log_dir = "logs"
+        max_log_size = 10 * 1024 * 1024  # 10MB
+        
         try:
-            log_dir = "logs"
-            if not os.path.exists(log_dir):
-                return
-                
-            current_date = datetime.utcnow().strftime("%Y%m%d")
             for filename in os.listdir(log_dir):
-                if filename.endswith(".log"):
-                    src_path = os.path.join(log_dir, filename)
-                    dst_path = os.path.join(log_dir, f"{filename}.{current_date}")
+                filepath = os.path.join(log_dir, filename)
+                if os.path.getsize(filepath) > max_log_size:
+                    # Archive old log
+                    archive_name = f"{filename}.{datetime.now().strftime('%Y%m%d')}"
+                    shutil.move(filepath, os.path.join(log_dir, archive_name))
                     
-                    if os.path.exists(src_path):
-                        os.rename(src_path, dst_path)
-                        
-            logger.info("Log rotation completed successfully")
+                    # Create new log file
+                    open(filepath, 'a').close()
+                    logger.info(f"Rotated log file: {filename}")
         except Exception as e:
-            logger.error(f"Log rotation failed: {str(e)}")
-            raise
+            logger.error(f"Log rotation error: {str(e)}")
 
-    async def manage_storage_quotas(self):
-        """Check and manage storage quotas"""
+    async def manage_storage_quotas(self) -> Dict[str, Any]:
+        """Manage storage quotas and cleanup"""
         try:
             results = {
-                "status": "ok",
                 "warnings": [],
                 "cleaned": 0
             }
             
-            # Check uploads directory size
-            uploads_dir = "uploads"
-            if os.path.exists(uploads_dir):
-                total_size = sum(
-                    os.path.getsize(os.path.join(dirpath, filename))
-                    for dirpath, _, filenames in os.walk(uploads_dir)
-                    for filename in filenames
-                )
-                
-                # Alert if total size exceeds 90% of quota (e.g., 10GB)
-                quota_limit = 10 * 1024 * 1024 * 1024  # 10GB in bytes
-                if total_size > (quota_limit * 0.9):
-                    warning_msg = f"Storage quota nearly reached: {total_size / quota_limit:.1%}"
-                    results["warnings"].append(warning_msg)
-                    await self._notify_resource_warning(warning_msg)
+            # Check and clean upload directories
+            upload_dirs = ["uploads/documents", "uploads/images"]
+            for directory in upload_dirs:
+                if os.path.exists(directory):
+                    total_size = sum(
+                        os.path.getsize(os.path.join(directory, f))
+                        for f in os.listdir(directory)
+                        if os.path.isfile(os.path.join(directory, f))
+                    ) / (1024 * 1024)  # Convert to MB
                     
-            # Check database size
-            db_stats = await db.db.command("dbStats")
-            db_size = db_stats["dataSize"] + db_stats["indexSize"]
+                    if total_size > settings.MAX_UPLOAD_DIR_SIZE_MB:
+                        results["warnings"].append(
+                            f"Upload directory {directory} exceeds size limit"
+                        )
             
-            # Alert if database size exceeds 90% of quota (e.g., 5GB)
-            db_quota = 5 * 1024 * 1024 * 1024  # 5GB in bytes
-            if db_size > (db_quota * 0.9):
-                warning_msg = f"Database quota nearly reached: {db_size / db_quota:.1%}"
-                results["warnings"].append(warning_msg)
-                await self._notify_resource_warning(warning_msg)
-            
-            # Clean up temporary uploads older than 24 hours
-            temp_dir = os.path.join("uploads", "temp")
-            if os.path.exists(temp_dir):
-                current_time = datetime.utcnow()
-                for file_name in os.listdir(temp_dir):
-                    file_path = os.path.join(temp_dir, file_name)
-                    file_age = datetime.fromtimestamp(os.path.getctime(file_path))
-                    
-                    if current_time - file_age > timedelta(hours=24):
-                        os.remove(file_path)
-                        results["cleaned"] += 1
+            # Clean up temporary files
+            temp_dirs = ["uploads/temp", "backups/temp"]
+            for directory in temp_dirs:
+                if os.path.exists(directory):
+                    # Remove files older than 24 hours
+                    cutoff = datetime.now() - timedelta(days=1)
+                    for filename in os.listdir(directory):
+                        filepath = os.path.join(directory, filename)
+                        if os.path.getctime(filepath) < cutoff.timestamp():
+                            os.remove(filepath)
+                            results["cleaned"] += 1
             
             return results
         except Exception as e:
-            logger.error(f"Error managing storage quotas: {str(e)}")
+            logger.error(f"Storage quota management error: {str(e)}")
             return {"error": str(e)}
 
-    async def _notify_resource_warning(self, message: str):
-        """Send notification for resource warnings"""
-        try:
-            # Get admin users
-            admin_users = await db.db["users"].find(
-                {"roles": "admin"}
-            ).to_list(None)
-
-            # Send notifications
-            for admin in admin_users:
-                await notifications.create_notification(
-                    user_id=str(admin["_id"]),
-                    title="System Resource Warning",
-                    message=message,
-                    notification_type="system_warning"
-                )
-        except Exception as e:
-            logger.error(f"Error sending resource warning: {str(e)}")
-
-    async def _get_directory_size(self, path: str) -> int:
-        """Get the total size of a directory in bytes"""
-        from pathlib import Path
-        return sum(f.stat().st_size for f in Path(path).glob('**/*') if f.is_file())
-
 maintenance = MaintenanceService()

app/services/notifications.py

@@ -3,7 +3,7 @@ from datetime import datetime
 from ..db.database import db
 from ..core.config import settings
 from ..utils.cache import cache
-from ..main import broadcast_notification
+from ..services.websocket import create_and_broadcast_notification
 
 class NotificationService:
     async def create_notification(
@@ -15,29 +15,13 @@ class NotificationService:
         data: Dict[str, Any] = None
     ):
         """Create and store a notification"""
-        notification = {
-            "user_id": user_id,
-            "title": title,
-            "message": message,
-            "type": notification_type,
-            "data": data,
-            "created_at": datetime.utcnow(),
-            "read": False
-        }
-        
-        # Store in database
-        await db.db["notifications"].insert_one(notification)
-        
-        # Broadcast to connected clients
-        await broadcast_notification({
-            "type": "notification",
-            "data": notification
-        })
-        
-        # Clear user's notification cache
-        await cache.delete_cache(f"user_notifications:{user_id}")
-        
-        return notification
+        return await create_and_broadcast_notification(
+            user_id=user_id,
+            title=title,
+            message=message,
+            notification_type=notification_type,
+            data=data
+        )
 
     async def get_user_notifications(
         self,

app/services/websocket.py

@@ -0,0 +1,63 @@
+from typing import List, Dict, Any
+from fastapi import WebSocket
+from datetime import datetime
+from ..db.database import db
+from ..utils.cache import cache
+
+# Store active WebSocket connections
+active_connections: List[WebSocket] = []
+
+async def connect(websocket: WebSocket):
+    """Accept a new WebSocket connection"""
+    await websocket.accept()
+    active_connections.append(websocket)
+
+async def disconnect(websocket: WebSocket):
+    """Remove a WebSocket connection"""
+    if websocket in active_connections:
+        active_connections.remove(websocket)
+
+async def broadcast_message(message: dict):
+    """Broadcast a message to all connected clients"""
+    disconnected = []
+    for connection in active_connections:
+        try:
+            await connection.send_json(message)
+        except:
+            disconnected.append(connection)
+    
+    # Clean up disconnected clients
+    for connection in disconnected:
+        await disconnect(connection)
+
+async def create_and_broadcast_notification(
+    user_id: str,
+    title: str,
+    message: str,
+    notification_type: str,
+    data: Dict[str, Any] = None
+) -> Dict[str, Any]:
+    """Create and broadcast a notification"""
+    notification = {
+        "user_id": user_id,
+        "title": title,
+        "message": message,
+        "type": notification_type,
+        "data": data,
+        "created_at": datetime.utcnow(),
+        "read": False
+    }
+    
+    # Store in database
+    await db.db["notifications"].insert_one(notification)
+    
+    # Broadcast to connected clients
+    await broadcast_message({
+        "type": "notification",
+        "data": notification
+    })
+    
+    # Clear user's notification cache
+    await cache.delete_cache(f"user_notifications:{user_id}")
+    
+    return notification

app/utils/tasks.py

@@ -1,167 +1,146 @@
-from datetime import datetime, timedelta
-from ..db.database import db
-from ..services.maintenance import maintenance
-from ..utils.logger import logger
-from ..services.websocket import broadcast_notification
-import asyncio
-from sqlalchemy import select
-from ..db.models import Event, User
-
-async def notify_users(user_ids: list, title: str, message: str, notification_type: str, data: dict = None):
-    """Send notification to users"""
-    for user_id in user_ids:
-        notification = {
-            "user_id": user_id,
-            "title": title,
-            "message": message,
-            "type": notification_type,
-            "data": data,
-            "created_at": datetime.utcnow(),
-            "read": False
-        }
-        
-        # Store in database
-        await db.db["notifications"].insert_one(notification)
-        
-        # Broadcast to connected clients
-        await broadcast_notification({
-            "type": "notification",
-            "data": notification
-        })
-
-async def check_event_reminders():
-    """Check and send event reminders"""
-    try:
-        now = datetime.utcnow()
-        # Find events happening soon that haven't sent reminders
-        stmt = select(Event).where(
-            Event.start_time > now,
-            Event.start_time <= now + timedelta(minutes=30),
-            Event.reminder_sent == False
-        )
-        
-        async with db.session() as session:
-            result = await session.execute(stmt)
-            upcoming_events = result.scalars().all()
-            
-            for event in upcoming_events:
-                # Get event owner
-                user_stmt = select(User).where(User.id == event.user_id)
-                user_result = await session.execute(user_stmt)
-                user = user_result.scalar_one_or_none()
-                
-                if user:
-                    # Send reminder to event owner and attendees
-                    reminder_users = [str(user.id)] + event.attendees
-                    await notify_users(
-                        user_ids=reminder_users,
-                        title=f"Event Reminder: {event.title}",
-                        message=f"Your event '{event.title}' starts in {event.reminder_minutes} minutes",
-                        notification_type="event_reminder",
-                        data={"event_id": str(event.id)}
-                    )
-                    
-                    # Mark reminder as sent
-                    event.reminder_sent = True
-                    await session.commit()
-
-    except Exception as e:
-        logger.error(f"Error in event reminder check: {str(e)}")
-
-async def cleanup_old_notifications():
-    """Clean up old notifications"""
-    try:
-        # Delete notifications older than 30 days
-        cutoff = datetime.utcnow() - timedelta(days=30)
-        result = await db.db["notifications"].delete_many({
-            "created_at": {"$lt": cutoff},
-            "read": True  # Only delete read notifications
-        })
-        logger.info(f"Cleaned up {result.deleted_count} old notifications")
-    except Exception as e:
-        logger.error(f"Error in cleanup_old_notifications: {str(e)}")
-
-async def perform_daily_maintenance():
-    """Perform daily system maintenance tasks"""
-    try:
-        # Clean up expired sessions
-        deleted_sessions = await maintenance.cleanup_expired_sessions()
-        logger.info(f"Cleaned up {deleted_sessions} expired sessions")
-
-        # Archive old data
-        archived = await maintenance.archive_old_data()
-        if archived:
-            logger.info(f"Archived data: {archived}")
-
-        # Check system health
-        health_data = await maintenance.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()
-        if "error" not in resources:
-            logger.info("System resource monitoring completed successfully")
-        else:
-            logger.error(f"Resource monitoring error: {resources['error']}")
-
-    except Exception as e:
-        logger.error(f"Error in daily maintenance: {str(e)}")
-
-async def perform_weekly_maintenance():
-    """Perform weekly system maintenance tasks"""
-    try:
-        # Perform database maintenance
-        await maintenance.perform_database_maintenance()
-        logger.info("Database maintenance completed successfully")
-
-        # Rotate log files
-        await maintenance.rotate_log_files()
-        logger.info("Log rotation completed successfully")
-
-        # Manage storage quotas
-        quota_results = await maintenance.manage_storage_quotas()
-        if quota_results.get("warnings"):
-            for warning in quota_results["warnings"]:
-                logger.warning(warning)
-        logger.info(f"Storage cleanup: removed {quota_results.get('cleaned', 0)} temporary files")
-
-    except Exception as e:
-        logger.error(f"Error in weekly maintenance: {str(e)}")
-
-async def run_periodic_tasks():
-    """Run periodic maintenance tasks"""
-    daily_maintenance_run = False
-    weekly_maintenance_run = False
-
-    while True:
-        try:
-            now = datetime.utcnow()
-
-            # Check event reminders every minute
-            await check_event_reminders()
-            
-            # Clean up old notifications daily
-            await cleanup_old_notifications()
-
-            # Run daily maintenance at 2 AM
-            if now.hour == 2 and not daily_maintenance_run:
-                await perform_daily_maintenance()
-                daily_maintenance_run = True
-            elif now.hour != 2:
-                daily_maintenance_run = False
-
-            # Run weekly maintenance on Sunday at 3 AM
-            if now.weekday() == 6 and now.hour == 3 and not weekly_maintenance_run:
-                await perform_weekly_maintenance()
-                weekly_maintenance_run = True
-            elif now.weekday() != 6 or now.hour != 3:
-                weekly_maintenance_run = False
-            
-            # Wait before next check
-            await asyncio.sleep(60)  # 1 minute
-        except Exception as e:
-            logger.error(f"Error in periodic tasks: {str(e)}")
-            await asyncio.sleep(60)  # Wait before retrying
+from datetime import datetime, timedelta
+from ..db.database import db
+from ..services.maintenance import maintenance
+from ..utils.logger import logger
+from ..services.websocket import create_and_broadcast_notification
+import asyncio
+from sqlalchemy import select
+from ..db.models import Event, User
+
+async def check_event_reminders():
+    """Check and send event reminders"""
+    try:
+        now = datetime.utcnow()
+        # Find events happening soon that haven't sent reminders
+        stmt = select(Event).where(
+            Event.start_time > now,
+            Event.start_time <= now + timedelta(minutes=30),
+            Event.reminder_sent == False
+        )
+        
+        async with db.session() as session:
+            result = await session.execute(stmt)
+            upcoming_events = result.scalars().all()
+            
+            for event in upcoming_events:
+                # Get event owner
+                user_stmt = select(User).where(User.id == event.user_id)
+                user_result = await session.execute(user_stmt)
+                user = user_result.scalar_one_or_none()
+                
+                if user:
+                    # Send reminder to event owner and attendees
+                    reminder_users = [str(user.id)] + event.attendees
+                    for user_id in reminder_users:
+                        await create_and_broadcast_notification(
+                            user_id=user_id,
+                            title=f"Event Reminder: {event.title}",
+                            message=f"Your event '{event.title}' starts in {event.reminder_minutes} minutes",
+                            notification_type="event_reminder",
+                            data={"event_id": str(event.id)}
+                        )
+                    
+                    # Mark reminder as sent
+                    event.reminder_sent = True
+                    await session.commit()
+
+    except Exception as e:
+        logger.error(f"Error in event reminder check: {str(e)}")
+
+async def cleanup_old_notifications():
+    """Clean up old notifications"""
+    try:
+        # Delete notifications older than 30 days
+        cutoff = datetime.utcnow() - timedelta(days=30)
+        result = await db.db["notifications"].delete_many({
+            "created_at": {"$lt": cutoff},
+            "read": True  # Only delete read notifications
+        })
+        logger.info(f"Cleaned up {result.deleted_count} old notifications")
+    except Exception as e:
+        logger.error(f"Error in cleanup_old_notifications: {str(e)}")
+
+async def perform_daily_maintenance():
+    """Perform daily system maintenance tasks"""
+    try:
+        # Clean up expired sessions
+        deleted_sessions = await maintenance.cleanup_expired_sessions()
+        logger.info(f"Cleaned up {deleted_sessions} expired sessions")
+
+        # Archive old data
+        archived = await maintenance.archive_old_data()
+        if archived:
+            logger.info(f"Archived data: {archived}")
+
+        # Check system health
+        health_data = await maintenance.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()
+        if "error" not in resources:
+            logger.info("System resource monitoring completed successfully")
+        else:
+            logger.error(f"Resource monitoring error: {resources['error']}")
+
+    except Exception as e:
+        logger.error(f"Error in daily maintenance: {str(e)}")
+
+async def perform_weekly_maintenance():
+    """Perform weekly system maintenance tasks"""
+    try:
+        # Perform database maintenance
+        await maintenance.perform_database_maintenance()
+        logger.info("Database maintenance completed successfully")
+
+        # Rotate log files
+        await maintenance.rotate_log_files()
+        logger.info("Log rotation completed successfully")
+
+        # Manage storage quotas
+        quota_results = await maintenance.manage_storage_quotas()
+        if quota_results.get("warnings"):
+            for warning in quota_results["warnings"]:
+                logger.warning(warning)
+        logger.info(f"Storage cleanup: removed {quota_results.get('cleaned', 0)} temporary files")
+
+    except Exception as e:
+        logger.error(f"Error in weekly maintenance: {str(e)}")
+
+async def run_periodic_tasks():
+    """Run periodic maintenance tasks"""
+    daily_maintenance_run = False
+    weekly_maintenance_run = False
+
+    while True:
+        try:
+            now = datetime.utcnow()
+
+            # Check event reminders every minute
+            await check_event_reminders()
+            
+            # Clean up old notifications daily
+            await cleanup_old_notifications()
+
+            # Run daily maintenance at 2 AM
+            if now.hour == 2 and not daily_maintenance_run:
+                await perform_daily_maintenance()
+                daily_maintenance_run = True
+            elif now.hour != 2:
+                daily_maintenance_run = False
+
+            # Run weekly maintenance on Sunday at 3 AM
+            if now.weekday() == 6 and now.hour == 3 and not weekly_maintenance_run:
+                await perform_weekly_maintenance()
+                weekly_maintenance_run = True
+            elif now.weekday() != 6 or now.hour != 3:
+                weekly_maintenance_run = False
+            
+            # Wait before next check
+            await asyncio.sleep(60)  # 1 minute
+        except Exception as e:
+            logger.error(f"Error in periodic tasks: {str(e)}")
+            await asyncio.sleep(60)  # Wait before retrying