Update app/utils/tasks.py

app/utils/tasks.py

@@ -1,139 +1,127 @@
-from datetime import datetime, timedelta
-from ..db.database import db
-from ..services.notifications import notifications
-from ..services.maintenance import maintenance
-from ..utils.logger import logger
-import asyncio
-
-async def check_event_reminders():
-    """Check and send event reminders"""
-    try:
-        now = datetime.utcnow()
-        reminder_window = now + timedelta(minutes=30)  # Check next 30 minutes
-
-        # Find events that need reminders
-        events = await db.db["events"].find({
-            "start_time": {
-                "$gte": now,
-                "$lte": reminder_window
-            },
-            "reminder_sent": {"$ne": True}
-        }).to_list(None)
-
-        for event in events:
-            # Send reminder to event 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"])}
-            )
-
-            # Send reminders to attendees
-            for attendee in event.get("attendees", []):
-                await notifications.create_notification(
-                    user_id=attendee,
-                    title=f"Event Reminder: {event['title']}",
-                    message=f"Event '{event['title']}' starts in {event['reminder_minutes']} minutes",
-                    notification_type="event_reminder",
-                    data={"event_id": str(event["_id"])}
-                )
-
-            # Mark reminder as sent
-            await db.db["events"].update_one(
-                {"_id": event["_id"]},
-                {"$set": {"reminder_sent": True}}
-            )
-
-    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_date = datetime.utcnow() - timedelta(days=30)
-        result = await db.db["notifications"].delete_many({
-            "created_at": {"$lt": cutoff_date}
-        })
-        logger.info(f"Cleaned up {result.deleted_count} old notifications")
-    except Exception as e:
-        logger.error(f"Error in notification cleanup: {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
-        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']}")
-
-    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 sqlalchemy.future import select
+from sqlalchemy import update, delete
+from sqlalchemy.orm import selectinload
+from ..db.database import AsyncSessionLocal
+from ..models.event import Event  # Your SQLAlchemy model for events
+from ..models.notification import Notification  # Your notification model
+from ..services.notifications import notifications
+from ..services.maintenance import maintenance
+from ..utils.logger import logger
+import asyncio
+
+async def check_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))
+
+            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,
+                        title=f"Event Reminder: {event.title}",
+                        message=f"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()
+
+    except Exception as e:
+        logger.error(f"Error in event reminder check: {str(e)}")
+
+async def cleanup_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.")
+    except Exception as e:
+        logger.error(f"Error in cleanup_old_notifications: {str(e)}")
+
+async def perform_daily_maintenance():
+    try:
+        deleted_sessions = await maintenance.cleanup_expired_sessions()
+        logger.info(f"Cleaned up {deleted_sessions} expired sessions")
+
+        archived = await maintenance.archive_old_data()
+        if archived:
+            logger.info(f"Archived data: {archived}")
+
+        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']}")
+
+        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():
+    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']}")
+    except Exception as e:
+        logger.error(f"Error in weekly maintenance: {str(e)}")
+
+async def run_periodic_tasks():
+    daily_maintenance_run = False
+    weekly_maintenance_run = False
+
+    while True:
+        try:
+            now = datetime.utcnow()
+
+            await check_event_reminders()
+            await cleanup_old_notifications()
+
+            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
+
+            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)
+        except Exception as e:
+            logger.error(f"Error in periodic tasks: {str(e)}")
+            await asyncio.sleep(60)