bk ser

app.py

@@ -39,6 +39,17 @@ async def lifespan(app: FastAPI):
     register_db_service_config()
     logger.info("✅ DB Service configured")
     
+    # Initialize Backup Service
+    from services.backup_service import initialize_backup_service
+    # Note: drive_service is already initialized globally, but re-initialized here as per instruction.
+    # The global drive_service is used later for download/upload.
+    local_drive_service = DriveService() 
+    backup_service = initialize_backup_service(
+        local_drive_service,
+        min_interval_seconds=30  # Minimum 30s between backups
+    )
+    logger.info("✅ Backup Service initialized")
+    
     # Register Auth Service configuration
     from services.auth_service import register_auth_service
     register_auth_service(
@@ -120,7 +131,9 @@ async def lifespan(app: FastAPI):
     
     # Shutdown: Upload DB to Drive
     logger.info("Shutdown: Uploading database to Google Drive...")
-    await drive_service.upload_db_async()
+    from services.backup_service import get_backup_service
+    backup_service = get_backup_service()
+    await backup_service.backup_async(force=True)  # Force backup on shutdown
     logger.info("Shutting down...")
 
 

routers/auth.py

@@ -200,7 +200,9 @@ async def google_auth(
     access_token = create_access_token(user.user_id, user.email, user.token_version)
     
     # Sync DB to Drive (Async)
-    background_tasks.add_task(drive_service.upload_db_async)
+    from services.backup_service import get_backup_service
+    backup_service = get_backup_service()
+    background_tasks.add_task(backup_service.backup_async)
     
     return AuthResponse(
         success=True,
@@ -336,6 +338,8 @@ async def logout(
     await db.commit()
     
     # Sync DB to Drive (Async)
-    background_tasks.add_task(drive_service.upload_db_async)
+    from services.backup_service import get_backup_service
+    backup_service = get_backup_service()
+    background_tasks.add_task(backup_service.backup_async)
     
     return {"success": True, "message": "Logged out successfully. All sessions invalidated."}

routers/payments.py

@@ -410,7 +410,9 @@ async def verify_payment(
         )
         
         # Sync DB to Drive (Async)
-        background_tasks.add_task(drive_service.upload_db_async)
+        from services.backup_service import get_backup_service
+        backup_service = get_backup_service()
+        background_tasks.add_task(backup_service.backup_async)
         
         return VerifyPaymentResponse(
             success=True,
@@ -505,7 +507,9 @@ async def razorpay_webhook(
                         )
                         
                         # Sync DB to Drive (Async)
-                        background_tasks.add_task(drive_service.upload_db_async)
+                        from services.backup_service import get_backup_service
+                        backup_service = get_backup_service()
+                        background_tasks.add_task(backup_service.backup_async)
         
         # Handle payment failed event
         elif event == "payment.failed":

services/backup_service/USAGE.md

@@ -0,0 +1,54 @@
+# Example usage in routes and app.py
+
+"""
+In app.py (startup):
+
+    from services.drive_service import DriveService
+    from services.backup_service import initialize_backup_service
+    
+    drive_service = DriveService()
+    backup_service = initialize_backup_service(
+        drive_service,
+        min_interval_seconds=30  # 30 seconds between backups
+    )
+
+In routes (e.g., payments.py):
+
+    from fastapi import BackgroundTasks
+    from services.backup_service import get_backup_service
+    
+    @router.post("/verify")
+    async def verify_payment(background_tasks: BackgroundTasks, ...):
+        # ... payment logic ...
+        
+        # Trigger backup in background (non-blocking)
+        backup_service = get_backup_service()
+        background_tasks.add_task(backup_service.backup_async)
+        
+        return {"success": True}
+
+In app.py (shutdown):
+
+    from services.backup_service import get_backup_service
+    
+    @app.on_event("shutdown")
+    async def shutdown():
+        backup_service = get_backup_service()
+        # Force backup on shutdown (ignore debouncing)
+        await backup_service.backup_async(force=True)
+
+How it works:
+
+1. User A pays → backup starts in background
+2. User B pays (1s later) → sees backup in progress, skips
+3. User C pays (2s later) → sees backup in progress, skips
+4. Backup completes (5s)
+5. User D pays (10s after first) → only 10s elapsed, skips (debounce)
+6. User E pays (35s after first) → 35s > 30s, new backup starts!
+
+Benefits:
+- ✅ No concurrent uploads
+- ✅ Response returns immediately
+- ✅ Minimal Drive API usage
+- ✅ No wasted resources
+"""

services/backup_service/__init__.py

@@ -0,0 +1,169 @@
+"""
+DB Backup Service - Intelligent database backup to Google Drive
+
+Features:
+- Async lock to prevent concurrent uploads
+- Debouncing to skip frequent uploads
+- Non-blocking background execution
+- Force option for critical backups (e.g., shutdown)
+"""
+
+import asyncio
+import logging
+from datetime import datetime, timedelta
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+class BackupService:
+    """
+    Intelligent database backup service.
+    
+    Prevents concurrent uploads and excessive backup frequency using:
+    - Async lock (one upload at a time)
+    - Debouncing (minimum interval between uploads)
+    - Non-blocking skip (don't wait if upload in progress)
+    """
+    
+    def __init__(
+        self,
+        drive_service,
+        min_interval_seconds: int = 30
+    ):
+        """
+        Initialize backup service.
+        
+        Args:
+            drive_service: DriveService instance for actual uploads
+            min_interval_seconds: Minimum seconds between uploads (debouncing)
+        """
+        self.drive_service = drive_service
+        self.min_interval = timedelta(seconds=min_interval_seconds)
+        
+        # Async lock to prevent concurrent uploads
+        self._upload_lock = asyncio.Lock()
+        
+        # Track last upload time for debouncing
+        self._last_upload_time: Optional[datetime] = None
+        
+        logger.info(f"BackupService initialized (min interval: {min_interval_seconds}s)")
+    
+    async def backup_async(self, force: bool = False) -> bool:
+        """
+        Backup database to Google Drive (async, non-blocking).
+        
+        Args:
+            force: If True, skip debouncing and upload immediately
+                  (useful for critical backups like shutdown)
+        
+        Returns:
+            True if backup was performed, False if skipped
+        """
+        # Non-blocking check: if upload already in progress, skip
+        if self._upload_lock.locked():
+            logger.info("📦 Backup already in progress, skipping this request")
+            return False
+        
+        # Debouncing check: skip if uploaded recently (unless forced)
+        if not force and self._last_upload_time:
+            time_since_last = datetime.utcnow() - self._last_upload_time
+            if time_since_last < self.min_interval:
+                logger.info(
+                    f"📦 Backup skipped - last backup was {time_since_last.seconds}s ago "
+                    f"(min interval: {self.min_interval.seconds}s)"
+                )
+                return False
+        
+        # Try to acquire lock (should succeed since we checked above)
+        async with self._upload_lock:
+            try:
+                # Perform actual upload using DriveService
+                loop = asyncio.get_event_loop()
+                await loop.run_in_executor(
+                    None,
+                    self.drive_service.upload_db
+                )
+                
+                # Update last upload time
+                self._last_upload_time = datetime.utcnow()
+                
+                logger.info("✅ Database backup completed successfully")
+                return True
+                
+            except Exception as e:
+                logger.error(f"❌ Database backup failed: {e}")
+                # Don't update last_upload_time on failure
+                # This allows retry on next request
+                return False
+    
+    def backup_sync(self, force: bool = False) -> bool:
+        """
+        Synchronous backup (for non-async contexts).
+        
+        Args:
+            force: If True, skip debouncing
+            
+        Returns:
+            True if backup was performed, False if skipped
+        """
+        # Check if we should skip (debouncing)
+        if not force and self._last_upload_time:
+            time_since_last = datetime.utcnow() - self._last_upload_time
+            if time_since_last < self.min_interval:
+                logger.info(
+                    f"📦 Backup skipped - last backup was {time_since_last.seconds}s ago"
+                )
+                return False
+        
+        try:
+            # Perform upload
+            self.drive_service.upload_db()
+            self._last_upload_time = datetime.utcnow()
+            logger.info("✅ Database backup completed successfully (sync)")
+            return True
+        except Exception as e:
+            logger.error(f"❌ Database backup failed (sync): {e}")
+            return False
+    
+    def get_stats(self) -> dict:
+        """Get backup statistics."""
+        return {
+            "last_backup_time": self._last_upload_time.isoformat() if self._last_upload_time else None,
+            "upload_in_progress": self._upload_lock.locked(),
+            "min_interval_seconds": self.min_interval.seconds,
+        }
+
+
+# Global instance (initialized in app startup)
+_backup_service: Optional[BackupService] = None
+
+
+def get_backup_service() -> BackupService:
+    """Get the global backup service instance."""
+    if _backup_service is None:
+        raise RuntimeError("BackupService not initialized. Call initialize_backup_service() first.")
+    return _backup_service
+
+
+def initialize_backup_service(drive_service, min_interval_seconds: int = 30) -> BackupService:
+    """
+    Initialize the global backup service instance.
+    
+    Args:
+        drive_service: DriveService instance
+        min_interval_seconds: Minimum seconds between backups
+        
+    Returns:
+        BackupService instance
+    """
+    global _backup_service
+    _backup_service = BackupService(drive_service, min_interval_seconds)
+    return _backup_service
+
+
+__all__ = [
+    'BackupService',
+    'get_backup_service',
+    'initialize_backup_service',
+]

services/gemini_service/job_processor.py

@@ -188,7 +188,9 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
                         job.completed_at = datetime.utcnow()
                         success = True
                         # Sync DB on success
-                        await self.drive_service.upload_db_async()
+                        from services.backup_service import get_backup_service
+                        backup_service = get_backup_service()
+                        await backup_service.backup_async()
                     else:
                         job.status = "failed"
                         job.error_message = "No video URL returned"
@@ -291,7 +293,9 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"status": "success", "type": "image_edit"}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self.drive_service.upload_db_async()
+        from services.backup_service import get_backup_service
+        backup_service = get_backup_service()
+        await backup_service.backup_async()
         return job
     
     async def _process_text(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -305,7 +309,9 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self.drive_service.upload_db_async()
+        from services.backup_service import get_backup_service
+        backup_service = get_backup_service()
+        await backup_service.backup_async()
         return job
     
     async def _process_analyze(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -320,7 +326,9 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self.drive_service.upload_db_async()
+        from services.backup_service import get_backup_service
+        backup_service = get_backup_service()
+        await backup_service.backup_async()
         return job
     
     async def _process_animation_prompt(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -335,7 +343,9 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self.drive_service.upload_db_async()
+        from services.backup_service import get_backup_service
+        backup_service = get_backup_service()
+        await backup_service.backup_async()
         return job