Add hourly Google Drive PDF ingestion

backend/app/config.py

@@ -24,6 +24,11 @@ class Settings(BaseSettings):
     JWT_REFRESH_EXPIRY_DAYS: int = 7
     GOOGLE_CLIENT_ID: str = ""
 
+    # Google Drive background sync
+    DRIVE_SYNC_ENABLED: bool = False
+    DRIVE_SYNC_INTERVAL_MINUTES: int = 60
+    GOOGLE_SERVICE_ACCOUNT_FILE: str = ""
+
     # ── File Upload ──────────────────────────────────────
     UPLOAD_DIR: str = "./data/uploads"
     MAX_UPLOAD_SIZE_MB: int = 20

backend/app/database.py

@@ -44,13 +44,15 @@ def _migrate_schema():
     for non-destructive changes such as new nullable columns.
     """
     inspector = inspect(engine)
-    existing_columns = {c["name"] for c in inspector.get_columns("users")}
-
     migrations = [
         ("users", "hf_token", "ALTER TABLE users ADD COLUMN hf_token VARCHAR(255)"),
+        ("documents", "drive_file_id", "ALTER TABLE documents ADD COLUMN drive_file_id VARCHAR(255)"),
+        ("documents", "drive_folder_id", "ALTER TABLE documents ADD COLUMN drive_folder_id VARCHAR(255)"),
+        ("documents", "drive_synced_at", "ALTER TABLE documents ADD COLUMN drive_synced_at TIMESTAMP"),
     ]
 
     for table, column, ddl in migrations:
+        existing_columns = {c["name"] for c in inspector.get_columns(table)}
         if column not in existing_columns:
             try:
                 with engine.begin() as conn:

backend/app/main.py

@@ -20,6 +20,7 @@ from app.config import get_settings
 from app.rate_limit import limiter
 from app.database import init_db, get_db
 from app.rag.vectorstore import get_chroma_client
+from app.scheduler import start_scheduler, stop_scheduler
 
 # Configure logging
 logging.basicConfig(
@@ -53,9 +54,12 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         logger.warning(f"Failed to pre-load embedding model: {e}")
 
+    start_scheduler()
+
     yield
 
     # ── Shutdown ─────────────────────────────────────
+    stop_scheduler()
     logger.info("Shutting down")
 
 

backend/app/models.py

@@ -28,6 +28,7 @@ class User(Base):
     documents = relationship("Document", back_populates="owner", cascade="all, delete-orphan")
     messages = relationship("ChatMessage", back_populates="user", cascade="all, delete-orphan")
     api_keys = relationship("ApiKey", back_populates="user", cascade="all, delete-orphan")
+    drive_connections = relationship("DriveConnection", back_populates="user", cascade="all, delete-orphan")
 
 
 class ApiKey(Base):
@@ -58,6 +59,9 @@ class Document(Base):
     error_message = Column(Text, nullable=True)
     uploaded_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
     summary = Column(Text, nullable=True)  # Optional summary of the document's content
+    drive_file_id = Column(String(255), unique=True, nullable=True, index=True)
+    drive_folder_id = Column(String(255), nullable=True, index=True)
+    drive_synced_at = Column(DateTime, nullable=True)
 
     # Relationships
     owner = relationship("User", back_populates="documents")
@@ -81,6 +85,22 @@ class ChatMessage(Base):
     shared_message = relationship("SharedMessage", back_populates="message", uselist=False, cascade="all, delete-orphan")
 
 
+class DriveConnection(Base):
+    __tablename__ = "drive_connections"
+
+    id = Column(String, primary_key=True, default=generate_uuid)
+    user_id = Column(String, ForeignKey("users.id"), nullable=False, index=True)
+    folder_id = Column(String(255), nullable=False, index=True)
+    credentials_json = Column(Text, nullable=True)
+    service_account_file = Column(String(500), nullable=True)
+    enabled = Column(Boolean, default=True, nullable=False)
+    last_synced_at = Column(DateTime, nullable=True)
+    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
+    updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
+
+    user = relationship("User", back_populates="drive_connections")
+
+
 class SharedMessage(Base):
     __tablename__ = "shared_messages"
 

backend/app/routes/documents.py

@@ -19,8 +19,8 @@ from app.models import User, Document
 from app.schemas import DocumentResponse, DocumentListResponse, DocumentStatusResponse
 from app.auth import get_current_user
 from app.config import get_settings
-from app.rag.chunker import chunk_document, get_page_count
-from app.rag.vectorstore import store_chunks, delete_document_chunks
+from app.rag.vectorstore import delete_document_chunks
+from app.services.document_ingestion import ingest_document
 from sqlalchemy import select
 logger = logging.getLogger(__name__)
 settings = get_settings()
@@ -113,96 +113,7 @@ async def validate_upload(file: UploadFile):
         pass
 
 
-def _ingest_document(document_id: str, filepath: str, original_name: str, user_id: str):
-    """
-    Process a document in the background: chunk document, generate embeddings, and store in ChromaDB,
-    calls document summary function, and update the database record.
-
-    This function is intended to be run as a background task. 
-    It creates its own database session, updates the
-    document status, extracts text, splits into chunks, generates embeddings,
-    stores everything in ChromaDB, calls summary function, updates the document record with page count,
-    chunk count, and summary, and marks the document as 'ready'. 
-    On failure, it sets status to 'failed' and records the error message.
-
-    Args:
-        document_id: Unique identifier of the document in the database.
-        filepath: Absolute or relative path to the uploaded file on disk.
-        original_name: original filename provided by the user (for logging and metadata).
-        user_id: Identifier of the user who owns the document.
-    
-    Returns:
-        None
-
-    Note:
-        This function does not raise exceptions to the caller;
-        all errors are logged and the database record is updated accordingly. 
-    """
-    from app.database import SessionLocal
-
-    db = SessionLocal()
-    try:
-        doc = db.query(Document).filter(Document.id == document_id).first()
-        if not doc:
-            logger.error(f"Document {document_id} not found for ingestion")
-            return
-
-        # Update status to processing
-        doc.status = "processing"
-        db.commit()
-
-        # Get page count
-        page_count = get_page_count(filepath)
-        doc.page_count = page_count
-
-        # Chunk the document
-        chunks = chunk_document(filepath)
-
-        if not chunks:
-            doc.status = "failed"
-            doc.error_message = "No text could be extracted from the document"
-            db.commit()
-            return
-
-        # Store embeddings in ChromaDB
-        chunk_count = store_chunks(
-            chunks=chunks,
-            document_id=document_id,
-            filename=original_name,
-            user_id=user_id,
-        )
-
-        # Generate summary and update document record
-        try:
-            from app.rag.summarizer import generate_document_summary
-
-            summary = generate_document_summary(filepath, max_sentences=2)
-            if summary:
-                doc.summary = summary
-                db.commit() # Update document record with summary                
-        except Exception as e:
-            logger.warning(f"Could not import summarizer for document {document_id}: {e}")
-            doc.summary = None
-
-        # Update document record
-        doc.chunk_count = chunk_count
-        doc.status = "ready"
-        db.commit()
-
-        logger.info(f"Document {document_id} ingested: {page_count} pages, {chunk_count} chunks")
-
-    except Exception as e:
-        logger.error(f"Ingestion error for {document_id}: {e}")
-        try:
-            doc = db.query(Document).filter(Document.id == document_id).first()
-            if doc:
-                doc.status = "failed"
-                doc.error_message = str(e)[:500]
-                db.commit()
-        except Exception:
-            pass
-    finally:
-        db.close()
+_ingest_document = ingest_document
 
 
 @router.post("/upload", response_model=DocumentResponse, status_code=status.HTTP_202_ACCEPTED)

backend/app/scheduler.py

@@ -0,0 +1,58 @@
+"""Application background scheduler."""
+import logging
+
+from app.config import get_settings
+from app.services.drive_sync import sync_drive_pdfs_with_session
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
+_scheduler = None
+
+
+def start_scheduler():
+    """Start recurring backend jobs."""
+    global _scheduler
+
+    if _scheduler and _scheduler.running:
+        return _scheduler
+
+    try:
+        from apscheduler.schedulers.background import BackgroundScheduler
+        from apscheduler.triggers.interval import IntervalTrigger
+    except ImportError:
+        if settings.DRIVE_SYNC_ENABLED:
+            logger.warning("Drive PDF sync enabled but APScheduler is not installed")
+        return None
+
+    _scheduler = BackgroundScheduler(timezone="UTC")
+
+    if settings.DRIVE_SYNC_ENABLED:
+        _scheduler.add_job(
+            sync_drive_pdfs_with_session,
+            trigger=IntervalTrigger(minutes=settings.DRIVE_SYNC_INTERVAL_MINUTES),
+            id="drive_pdf_sync",
+            name="Hourly Google Drive PDF sync",
+            replace_existing=True,
+            max_instances=1,
+            coalesce=True,
+            misfire_grace_time=300,
+        )
+        logger.info(
+            "Drive PDF sync scheduled every %s minutes",
+            settings.DRIVE_SYNC_INTERVAL_MINUTES,
+        )
+    else:
+        logger.info("Drive PDF sync disabled")
+
+    _scheduler.start()
+    return _scheduler
+
+
+def stop_scheduler():
+    """Stop recurring backend jobs."""
+    global _scheduler
+
+    if _scheduler and _scheduler.running:
+        _scheduler.shutdown(wait=False)
+        logger.info("Background scheduler stopped")
+    _scheduler = None

backend/app/services/__init__.py

@@ -0,0 +1 @@
+"""Backend service layer helpers."""

backend/app/services/document_ingestion.py

@@ -0,0 +1,80 @@
+"""Reusable document ingestion pipeline."""
+import logging
+
+from app.models import Document
+from app.rag.chunker import chunk_document, get_page_count
+from app.rag.vectorstore import store_chunks
+
+logger = logging.getLogger(__name__)
+
+
+def ingest_document(document_id: str, filepath: str, original_name: str, user_id: str):
+    """
+    Process a document: chunk it, generate embeddings, store vectors, summarize,
+    and update the database record.
+    """
+    from app.database import SessionLocal
+
+    db = SessionLocal()
+    try:
+        doc = db.query(Document).filter(Document.id == document_id).first()
+        if not doc:
+            logger.error("Document %s not found for ingestion", document_id)
+            return
+
+        doc.status = "processing"
+        db.commit()
+
+        page_count = get_page_count(filepath)
+        doc.page_count = page_count
+
+        chunks = chunk_document(filepath)
+
+        if not chunks:
+            doc.status = "failed"
+            doc.error_message = "No text could be extracted from the document"
+            db.commit()
+            return
+
+        chunk_count = store_chunks(
+            chunks=chunks,
+            document_id=document_id,
+            filename=original_name,
+            user_id=user_id,
+        )
+
+        try:
+            from app.rag.summarizer import generate_document_summary
+
+            summary = generate_document_summary(filepath, max_sentences=2)
+            if summary:
+                doc.summary = summary
+                db.commit()
+        except Exception as e:
+            logger.warning("Could not generate summary for document %s: %s", document_id, e)
+            doc.summary = None
+
+        doc.chunk_count = chunk_count
+        doc.status = "ready"
+        doc.error_message = None
+        db.commit()
+
+        logger.info(
+            "Document %s ingested: %s pages, %s chunks",
+            document_id,
+            page_count,
+            chunk_count,
+        )
+
+    except Exception as e:
+        logger.error("Ingestion error for %s: %s", document_id, e)
+        try:
+            doc = db.query(Document).filter(Document.id == document_id).first()
+            if doc:
+                doc.status = "failed"
+                doc.error_message = str(e)[:500]
+                db.commit()
+        except Exception:
+            logger.exception("Failed to mark document %s as failed", document_id)
+    finally:
+        db.close()

backend/app/services/drive_sync.py

@@ -0,0 +1,188 @@
+"""Google Drive PDF discovery and ingestion service."""
+import io
+import json
+import logging
+import os
+import threading
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy.orm import Session
+
+from app.config import get_settings
+from app.models import Document, DriveConnection
+from app.services.document_ingestion import ingest_document
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
+_sync_lock = threading.Lock()
+
+PDF_MIME_TYPE = "application/pdf"
+DRIVE_SCOPES = ["https://www.googleapis.com/auth/drive.readonly"]
+
+
+def _build_drive_service(connection: DriveConnection):
+    """Build a Google Drive API client for a saved connection."""
+    try:
+        from google.oauth2.credentials import Credentials
+        from google.oauth2 import service_account
+        from googleapiclient.discovery import build
+    except ImportError as exc:
+        raise RuntimeError(
+            "Google Drive sync requires google-api-python-client and google-auth"
+        ) from exc
+
+    if connection.credentials_json:
+        credentials = Credentials.from_authorized_user_info(
+            json.loads(connection.credentials_json),
+            scopes=DRIVE_SCOPES,
+        )
+    else:
+        service_account_file = (
+            connection.service_account_file or settings.GOOGLE_SERVICE_ACCOUNT_FILE
+        )
+        if not service_account_file:
+            raise RuntimeError("Drive connection has no OAuth credentials or service account file")
+        credentials = service_account.Credentials.from_service_account_file(
+            service_account_file,
+            scopes=DRIVE_SCOPES,
+        )
+
+    return build("drive", "v3", credentials=credentials, cache_discovery=False)
+
+
+def _list_pdf_files(service, folder_id: str) -> list[dict]:
+    """Return PDF files directly inside a Drive folder."""
+    files: list[dict] = []
+    page_token = None
+    query = (
+        f"'{folder_id}' in parents and "
+        f"mimeType = '{PDF_MIME_TYPE}' and trashed = false"
+    )
+
+    while True:
+        response = (
+            service.files()
+            .list(
+                q=query,
+                spaces="drive",
+                fields="nextPageToken, files(id, name, mimeType, size, modifiedTime)",
+                pageToken=page_token,
+            )
+            .execute()
+        )
+        files.extend(response.get("files", []))
+        page_token = response.get("nextPageToken")
+        if not page_token:
+            return files
+
+
+def _download_drive_file(service, file_id: str, destination: str):
+    """Download a Drive file to disk."""
+    from googleapiclient.http import MediaIoBaseDownload
+
+    request = service.files().get_media(fileId=file_id)
+    with io.FileIO(destination, "wb") as file_handle:
+        downloader = MediaIoBaseDownload(file_handle, request)
+        done = False
+        while not done:
+            _, done = downloader.next_chunk()
+
+
+def sync_drive_pdfs(db: Session) -> dict:
+    """
+    Discover new PDFs from enabled Drive connections, download them, and ingest them.
+    """
+    if not _sync_lock.acquire(blocking=False):
+        logger.info("Drive sync skipped because a previous run is still active")
+        return {"connections": 0, "discovered": 0, "ingested": 0, "skipped": 0, "failed": 0}
+
+    stats = {"connections": 0, "discovered": 0, "ingested": 0, "skipped": 0, "failed": 0}
+    try:
+        connections = (
+            db.query(DriveConnection)
+            .filter(DriveConnection.enabled.is_(True))
+            .all()
+        )
+        stats["connections"] = len(connections)
+
+        for connection in connections:
+            try:
+                service = _build_drive_service(connection)
+                files = _list_pdf_files(service, connection.folder_id)
+                stats["discovered"] += len(files)
+
+                for drive_file in files:
+                    existing = (
+                        db.query(Document)
+                        .filter(Document.drive_file_id == drive_file["id"])
+                        .first()
+                    )
+                    if existing:
+                        stats["skipped"] += 1
+                        continue
+
+                    user_dir = os.path.join(settings.UPLOAD_DIR, connection.user_id)
+                    os.makedirs(user_dir, exist_ok=True)
+                    stored_filename = f"{uuid.uuid4().hex}.pdf"
+                    filepath = os.path.join(user_dir, stored_filename)
+
+                    document = Document(
+                        user_id=connection.user_id,
+                        filename=stored_filename,
+                        original_name=drive_file.get("name") or stored_filename,
+                        file_size=int(drive_file.get("size") or 0),
+                        status="pending",
+                        drive_file_id=drive_file["id"],
+                        drive_folder_id=connection.folder_id,
+                        drive_synced_at=datetime.now(timezone.utc),
+                    )
+                    db.add(document)
+                    db.commit()
+                    db.refresh(document)
+
+                    try:
+                        _download_drive_file(service, drive_file["id"], filepath)
+                        document.file_size = Path(filepath).stat().st_size
+                        db.commit()
+                        ingest_document(
+                            document_id=document.id,
+                            filepath=filepath,
+                            original_name=document.original_name,
+                            user_id=connection.user_id,
+                        )
+                        stats["ingested"] += 1
+                    except Exception as exc:
+                        logger.exception(
+                            "Drive ingestion failed for file %s",
+                            drive_file.get("id"),
+                        )
+                        db.refresh(document)
+                        document.status = "failed"
+                        document.error_message = str(exc)[:500]
+                        db.commit()
+                        stats["failed"] += 1
+
+                connection.last_synced_at = datetime.now(timezone.utc)
+                connection.updated_at = datetime.now(timezone.utc)
+                db.commit()
+            except Exception:
+                logger.exception("Drive sync failed for connection %s", connection.id)
+                stats["failed"] += 1
+
+        logger.info("Drive sync complete: %s", stats)
+        return stats
+    finally:
+        _sync_lock.release()
+
+
+def sync_drive_pdfs_with_session() -> dict:
+    """Run Drive sync with an owned database session."""
+    from app.database import SessionLocal
+
+    db = SessionLocal()
+    try:
+        return sync_drive_pdfs(db)
+    finally:
+        db.close()

backend/requirements.txt

@@ -15,6 +15,8 @@ pyjwt
 passlib[bcrypt]
 python-dotenv
 google-auth
+google-api-python-client
+APScheduler
 
 # Config
 pydantic-settings

backend/tests/test_drive_sync.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+
+from app.models import Document, DriveConnection
+from app.services import drive_sync
+
+
+def test_drive_sync_ingests_new_pdf_and_skips_existing(db_session, user, tmp_path, monkeypatch):
+    monkeypatch.setattr(drive_sync.settings, "UPLOAD_DIR", str(tmp_path))
+    monkeypatch.setattr(drive_sync, "_build_drive_service", lambda connection: object())
+    monkeypatch.setattr(
+        drive_sync,
+        "_list_pdf_files",
+        lambda service, folder_id: [
+            {"id": "drive-file-1", "name": "Guide.pdf", "size": "0"},
+        ],
+    )
+
+    def fake_download(service, file_id, destination):
+        Path(destination).write_bytes(b"%PDF-1.4\n")
+
+    def fake_ingest(document_id, filepath, original_name, user_id):
+        document = db_session.query(Document).filter(Document.id == document_id).one()
+        document.status = "ready"
+        document.page_count = 1
+        document.chunk_count = 1
+        db_session.commit()
+
+    monkeypatch.setattr(drive_sync, "_download_drive_file", fake_download)
+    monkeypatch.setattr(drive_sync, "ingest_document", fake_ingest)
+
+    connection = DriveConnection(user_id=user.id, folder_id="folder-1")
+    db_session.add(connection)
+    db_session.commit()
+
+    first_stats = drive_sync.sync_drive_pdfs(db_session)
+    second_stats = drive_sync.sync_drive_pdfs(db_session)
+
+    document = db_session.query(Document).filter(Document.drive_file_id == "drive-file-1").one()
+    assert first_stats["ingested"] == 1
+    assert second_stats["skipped"] == 1
+    assert document.status == "ready"
+    assert document.original_name == "Guide.pdf"
+    assert document.drive_folder_id == "folder-1"