Update app/utils/tasks.py

app/utils/tasks.py

@@ -1,104 +1,138 @@
 from datetime import datetime, timedelta
-from sqlalchemy.future import select
-from sqlalchemy import update, delete
-from sqlalchemy.orm import selectinload
-from ..db.database import AsyncSessionLocal
-from app.models.event import Event
-from ..models.notification import Notification  # Your notification model
-from ..services.notifications import notifications
+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:
-        async with AsyncSessionLocal() as session:
-            now = datetime.utcnow()
-            reminder_window = now + timedelta(minutes=30)
-
-            # Fetch events within the window that haven’t been reminded yet
-            stmt = select(Event).where(
-                Event.start_time >= now,
-                Event.start_time <= reminder_window,
-                Event.reminder_sent == False
-            ).options(selectinload(Event.attendees))
-
+        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)
-            events = result.scalars().all()
-
-            for event in events:
-                # Notify the creator
-                await notifications.create_notification(
-                    user_id=event.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)}
-                )
-
-                # Notify attendees
-                for attendee in event.attendees:
-                    await notifications.create_notification(
-                        user_id=attendee.user_id,
+            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"Event '{event.title}' starts in {event.reminder_minutes} minutes",
+                        message=f"Your event '{event.title}' starts in {event.reminder_minutes} minutes",
                         notification_type="event_reminder",
                         data={"event_id": str(event.id)}
                     )
-
-                # Mark reminder sent
-                event.reminder_sent = True
-
-            await session.commit()
+                    
+                    # 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:
-        async with AsyncSessionLocal() as session:
-            cutoff = datetime.utcnow() - timedelta(days=30)
-            stmt = delete(Notification).where(Notification.created_at < cutoff)
-            result = await session.execute(stmt)
-            await session.commit()
-            logger.info(f"Cleaned up old notifications.")
+        # 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:
-        db_maintenance = await maintenance.perform_database_maintenance()
-        if "error" not in db_maintenance:
-            logger.info("Database maintenance completed successfully")
-        else:
-            logger.error(f"Database maintenance error: {db_maintenance['error']}")
+        # 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
 
@@ -106,22 +140,28 @@ async def run_periodic_tasks():
         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
-
-            await asyncio.sleep(60)
+            
+            # 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)
+            await asyncio.sleep(60)  # Wait before retrying