drive sync

routers/payments.py

@@ -14,7 +14,7 @@ import uuid
 from datetime import datetime
 from typing import List, Optional
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status, BackgroundTasks
 from pydantic import BaseModel
 from sqlalchemy import select, desc, func
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -22,6 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from core.database import get_db
 from core.models import User, PaymentTransaction
 from dependencies import get_current_user
+from services.drive_service import DriveService
 from services.razorpay_service import (
     RazorpayService,
     RazorpayConfigError,
@@ -37,6 +38,7 @@ from services.razorpay_service import (
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/payments", tags=["payments"])
+drive_service = DriveService()
 
 
 # =============================================================================
@@ -314,6 +316,7 @@ async def create_order(
 @router.post("/verify", response_model=VerifyPaymentResponse)
 async def verify_payment(
     request: VerifyPaymentRequest,
+    background_tasks: BackgroundTasks,
     user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db)
 ):
@@ -403,6 +406,9 @@ async def verify_payment(
             signature=request.razorpay_signature
         )
         
+        # Sync DB to Drive (Async)
+        background_tasks.add_task(drive_service.upload_db)
+        
         return VerifyPaymentResponse(
             success=True,
             message="Payment successful! Credits added.",
@@ -422,6 +428,7 @@ async def verify_payment(
 @router.post("/webhook/razorpay")
 async def razorpay_webhook(
     request: Request,
+    background_tasks: BackgroundTasks,
     db: AsyncSession = Depends(get_db)
 ):
     """
@@ -493,6 +500,9 @@ async def razorpay_webhook(
                             source="webhook",
                             db=db
                         )
+                        
+                        # Sync DB to Drive (Async)
+                        background_tasks.add_task(drive_service.upload_db)
         
         # Handle payment failed event
         elif event == "payment.failed":

services/gemini_job_worker.py

@@ -18,6 +18,8 @@ from services.priority_worker_pool import (
     get_interval_for_priority
 )
 from services.gemini_service import GeminiService
+from services.drive_service import DriveService
+import asyncio
 
 logger = logging.getLogger(__name__)
 
@@ -40,6 +42,17 @@ def get_priority_for_job_type(job_type: str) -> str:
 class GeminiJobProcessor(JobProcessor[GeminiJob]):
     """Processes Gemini AI jobs (text, image, video generation) with round-robin API keys."""
     
+    def __init__(self):
+        self.drive_service = DriveService()
+        
+    async def _sync_db_to_drive(self):
+        """Sync database to Google Drive in a separate thread."""
+        try:
+            # Run synchronous upload_db in a thread to avoid blocking the event loop
+            await asyncio.to_thread(self.drive_service.upload_db)
+        except Exception as e:
+            logger.error(f"Failed to sync DB to Drive after job completion: {e}")
+
     async def _get_service_with_key(self, session: AsyncSession) -> tuple:
         """Get a GeminiService with the least-used API key."""
         from services.api_key_manager import get_least_used_key
@@ -182,6 +195,8 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
                         job.error_message = None  # Clear any previous error
                         job.completed_at = datetime.utcnow()
                         success = True
+                        # Sync DB on success
+                        await self._sync_db_to_drive()
                     else:
                         job.status = "failed"
                         job.error_message = "No video URL returned"
@@ -283,6 +298,8 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         # Don't save full base64 image to api_response
         job.api_response = {"status": "success", "type": "image_edit"}
         job.completed_at = datetime.utcnow()
+        # Sync DB on success
+        await self._sync_db_to_drive()
         return job
     
     async def _process_text(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -295,6 +312,8 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.output_data = {"text": result}
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
+        # Sync DB on success
+        await self._sync_db_to_drive()
         return job
     
     async def _process_analyze(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -308,6 +327,8 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.output_data = {"analysis": result}
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
+        # Sync DB on success
+        await self._sync_db_to_drive()
         return job
     
     async def _process_animation_prompt(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -321,6 +342,8 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.output_data = {"prompt": result}
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
+        # Sync DB on success
+        await self._sync_db_to_drive()
         return job