QA version persionality

.gitignore

@@ -59,7 +59,6 @@ out/
 tests/
 
 Admin_bot/
-
 Pix-Agent/
 
 # Hugging Face Spaces
@@ -81,3 +80,6 @@ Thumbs.db
 main.py
 
 test/
+
+/tmp
+/docs/

README.md

@@ -416,4 +416,145 @@ Lịch sử hội thoại người dùng được lưu trong queue riêng với
 
 ## Tác giả
 
-- **PIX Project Team** 
+- **PIX Project Team**
+
+# PixAgent PDF Processing
+
+This README provides instructions for the PDF processing functionality in PixAgent, including uploading PDF documents, managing vector embeddings, and deleting documents.
+
+## API Endpoints
+
+### Health Check
+
+```
+GET /health
+GET /pdf/health
+```
+
+Verify the API is running and the connection to databases (MongoDB, PostgreSQL, Pinecone) is established.
+
+### Upload PDF
+
+```
+POST /pdf/upload
+```
+
+**Parameters:**
+- `file`: The PDF file to upload (multipart/form-data)
+- `namespace`: The namespace to store vectors in (default: "Default")
+- `mock_mode`: Set to "true" or "false" (default: "false")
+- `vector_database_id`: The ID of the vector database to use (required for real mode)
+- `document_id`: Optional custom document ID (if not provided, a UUID will be generated)
+
+**Example Python Request:**
+```python
+import requests
+import uuid
+
+document_id = str(uuid.uuid4())
+files = {'file': open('your_document.pdf', 'rb')}
+response = requests.post(
+    'http://localhost:8000/pdf/upload',
+    files=files,
+    data={
+        'namespace': 'my-namespace',
+        'mock_mode': 'false',
+        'vector_database_id': '9',
+        'document_id': document_id
+    }
+)
+print(f'Status: {response.status_code}')
+print(f'Response: {response.json()}')
+```
+
+### List Documents
+
+```
+GET /pdf/documents
+```
+
+**Parameters:**
+- `namespace`: The namespace to retrieve documents from
+- `vector_database_id`: The ID of the vector database to use
+
+**Example Python Request:**
+```python
+import requests
+
+response = requests.get(
+    'http://localhost:8000/pdf/documents',
+    params={
+        'namespace': 'my-namespace',
+        'vector_database_id': '9'
+    }
+)
+print(f'Status: {response.status_code}')
+print(f'Documents: {response.json()}')
+```
+
+### Delete Document
+
+```
+DELETE /pdf/document
+```
+
+**Parameters:**
+- `document_id`: The ID of the document to delete
+- `namespace`: The namespace containing the document
+- `vector_database_id`: The ID of the vector database
+
+**Example Python Request:**
+```python
+import requests
+
+response = requests.delete(
+    'http://localhost:8000/pdf/document',
+    params={
+        'document_id': 'your-document-id',
+        'namespace': 'my-namespace',
+        'vector_database_id': '9'
+    }
+)
+print(f'Status: {response.status_code}')
+print(f'Result: {response.json()}')
+```
+
+### List Available Vector Databases
+
+```
+GET /postgres/vector-databases
+```
+
+**Example Python Request:**
+```python
+import requests
+
+response = requests.get('http://localhost:8000/postgres/vector-databases')
+vector_dbs = response.json()
+print(f'Available vector databases: {vector_dbs}')
+```
+
+## PDF Processing and Vector Embedding
+
+The system processes PDFs in the following steps:
+
+1. **Text Extraction**: Uses `PyPDFLoader` from LangChain to extract text from the PDF.
+2. **Text Chunking**: Splits the text into manageable chunks using `RecursiveCharacterTextSplitter` with a chunk size of 1000 characters and 100 character overlap.
+3. **Embedding Creation**: Uses Google's Gemini embedding model (`models/embedding-001`) to create embeddings for each text chunk.
+4. **Dimension Adjustment**: Ensures the embedding dimensions match the Pinecone index requirements:
+   - If Gemini produces 768-dim embeddings and Pinecone expects 1536-dim, each value is duplicated.
+   - For other mismatches, appropriate padding or truncation is applied.
+5. **Vector Storage**: Uploads the embeddings to Pinecone in the specified namespace.
+
+## Notes
+
+- **Mock Mode**: When `mock_mode` is set to "true", the system simulates the PDF processing without actually creating or storing embeddings.
+- **Namespace Handling**: When using a vector database ID, the namespace is automatically formatted as `vdb-{vector_database_id}`.
+- **Error Handling**: The system validates vector dimensions and handles errors appropriately, with detailed logging.
+- **PDF Storage**: Processed PDFs are stored in the `pdf_storage` directory with the document ID as the filename.
+
+## Troubleshooting
+
+- **Dimension Mismatch Error**: If you receive an error about vector dimensions not matching Pinecone index configuration, check that the embedding model and Pinecone index dimensions are compatible. The system will attempt to adjust dimensions but may encounter limits.
+- **Connection Issues**: Verify that MongoDB, PostgreSQL, and Pinecone credentials are correctly configured in the environment variables.
+- **Processing Failures**: Check the `pdf_api_debug.log` file for detailed error messages and processing information. 

app.py

@@ -6,6 +6,11 @@ import os
 import sys
 import logging
 from dotenv import load_dotenv
+from fastapi.responses import JSONResponse, PlainTextResponse
+from fastapi.staticfiles import StaticFiles
+import time
+import uuid
+import traceback
 
 # Cấu hình logging
 logging.basicConfig(
@@ -83,6 +88,7 @@ try:
     from app.api.rag_routes import router as rag_router
     from app.api.websocket_routes import router as websocket_router
     from app.api.pdf_routes import router as pdf_router
+    from app.api.pdf_websocket import router as pdf_websocket_router
     
     # Import middlewares
     from app.utils.middleware import RequestLoggingMiddleware, ErrorHandlingMiddleware, DatabaseCheckMiddleware
@@ -93,6 +99,8 @@ try:
     # Import cache
     from app.utils.cache import get_cache
     
+    logger.info("Successfully imported all routers and modules")
+    
 except ImportError as e:
     logger.error(f"Error importing routes or middlewares: {e}")
     raise
@@ -129,6 +137,14 @@ app.include_router(postgresql_router)
 app.include_router(rag_router)
 app.include_router(websocket_router)
 app.include_router(pdf_router)
+app.include_router(pdf_websocket_router)
+
+# Log all registered routes
+logger.info("Registered API routes:")
+for route in app.routes:
+    if hasattr(route, "path") and hasattr(route, "methods"):
+        methods = ",".join(route.methods)
+        logger.info(f"  {methods:<10} {route.path}")
 
 # Root endpoint
 @app.get("/")
@@ -235,8 +251,37 @@ if DEBUG:
                 "history_cache_ttl": os.getenv("HISTORY_CACHE_TTL", "3600"),
             }
         }
+    
+    @app.get("/debug/websocket-routes")
+    def debug_websocket_routes():
+        """Hiển thị thông tin về các WebSocket route (chỉ trong chế độ debug)"""
+        ws_routes = []
+        for route in app.routes:
+            if "websocket" in str(route.__class__).lower():
+                ws_routes.append({
+                    "path": route.path,
+                    "name": route.name,
+                    "endpoint": str(route.endpoint)
+                })
+        return {
+            "websocket_routes": ws_routes,
+            "total_count": len(ws_routes)
+        }
+        
+    @app.get("/debug/mock-status")
+    def debug_mock_status():
+        """Display current mock mode settings"""
+        # Import was: from app.api.pdf_routes import USE_MOCK_MODE
+        # We've disabled mock mode
+        
+        return {
+            "mock_mode": False,  # Disabled - using real database
+            "mock_env_variable": os.getenv("USE_MOCK_MODE", "false"),
+            "debug_mode": DEBUG
+        }
+
 
 # Run the app with uvicorn when executed directly
 if __name__ == "__main__":
-    port = int(os.environ.get("PORT", 8000))
-    uvicorn.run("app:app", host="0.0.0.0", port=port, reload=DEBUG) 
+    port = int(os.environ.get("PORT", 7860))
+    uvicorn.run("app:app", host="0.0.0.0", port=port, reload=DEBUG) 

app/__init__.py

@@ -10,16 +10,11 @@ import os
 # Thêm thư mục gốc vào sys.path
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-try:
-    # Sửa lại cách import đúng - 'app.py' không phải là module hợp lệ
-    # 'app' là tên module, '.py' là phần mở rộng tệp
-    from app import app
-except ImportError:
-    # Thử cách khác nếu import trực tiếp không hoạt động
-    import importlib.util
-    spec = importlib.util.spec_from_file_location("app_module", 
-                                                 os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 
-                                                             "app.py"))
-    app_module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(app_module)
-    app = app_module.app 
+# Sử dụng importlib để tránh circular import
+import importlib.util
+spec = importlib.util.spec_from_file_location("app_module", 
+                                             os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 
+                                                         "app.py"))
+app_module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(app_module)
+app = app_module.app 

app/api/mongodb_routes.py

@@ -74,8 +74,8 @@ async def create_session(session: SessionCreate, response: Response):
             created_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         )
         
-        # Kiểm tra nếu session cần gửi thông báo (response bắt đầu bằng "I don't know")
-        if session.response and session.response.strip().lower().startswith("i don't know"):
+        # Kiểm tra nếu session cần gửi thông báo (response bắt đầu bằng "I'm sorry")
+        if session.response and session.response.strip().lower().startswith("i'm sorry"):
             # Gửi thông báo qua WebSocket
             try:
                 notification_data = {
@@ -93,7 +93,7 @@ async def create_session(session: SessionCreate, response: Response):
                 
                 # Khởi tạo task để gửi thông báo - sử dụng asyncio.create_task để đảm bảo không block quá trình chính
                 asyncio.create_task(send_notification(notification_data))
-                logger.info(f"Notification queued for session {session.session_id} - response starts with 'I don't know'")
+                logger.info(f"Notification queued for session {session.session_id} - response starts with 'I'm sorry'")
             except Exception as e:
                 logger.error(f"Error queueing notification: {e}")
                 # Không dừng xử lý chính khi gửi thông báo thất bại

app/api/pdf_routes.py

@@ -1,12 +1,23 @@
 import os
 import shutil
 import uuid
-from fastapi import APIRouter, UploadFile, File, Form, HTTPException, BackgroundTasks
+import sys
+import traceback
+from fastapi import APIRouter, UploadFile, File, Form, HTTPException, BackgroundTasks, Depends, Query
 from fastapi.responses import JSONResponse
 from typing import Optional, List, Dict, Any
+from sqlalchemy.orm import Session
+import os.path
+import logging
+import tempfile
+import time
+import json
+from datetime import datetime
 
 from app.utils.pdf_processor import PDFProcessor
 from app.models.pdf_models import PDFResponse, DeleteDocumentRequest, DocumentsListResponse
+from app.database.postgresql import get_db
+from app.database.models import VectorDatabase, Document, VectorStatus, ApiKey, DocumentContent
 from app.api.pdf_websocket import (
     send_pdf_upload_started, 
     send_pdf_upload_progress, 
@@ -17,21 +28,156 @@ from app.api.pdf_websocket import (
     send_pdf_delete_failed
 )
 
-# Khởi tạo router
+# Setup logger
+logger = logging.getLogger(__name__)
+
+# Add a stream handler for PDF debug logging
+pdf_debug_logger = logging.getLogger("pdf_debug_api")
+pdf_debug_logger.setLevel(logging.DEBUG)
+
+# Check if a stream handler already exists, add one if not
+if not any(isinstance(h, logging.StreamHandler) for h in pdf_debug_logger.handlers):
+    stream_handler = logging.StreamHandler(sys.stdout)
+    stream_handler.setLevel(logging.INFO)
+    pdf_debug_logger.addHandler(stream_handler)
+
+# Initialize router
 router = APIRouter(
     prefix="/pdf",
     tags=["PDF Processing"],
 )
 
-# Thư mục lưu file tạm - sử dụng /tmp để tránh lỗi quyền truy cập
-TEMP_UPLOAD_DIR = "/tmp/uploads/temp"
-STORAGE_DIR = "/tmp/uploads/pdfs"
+# Constants - Use system temp directory instead of creating our own
+TEMP_UPLOAD_DIR = tempfile.gettempdir()
+STORAGE_DIR = tempfile.gettempdir()  # Also use system temp for storage
+
+USE_MOCK_MODE = False  # Disabled - using real database with improved connection handling
+logger.info(f"PDF API starting with USE_MOCK_MODE={USE_MOCK_MODE}")
+
+# Helper function to log with timestamp
+def log_with_timestamp(message: str, level: str = "info", error: Exception = None):
+    """Add timestamps to log messages and log to the PDF debug logger if available"""
+    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    full_message = f"{timestamp} - {message}"
+    
+    if level.lower() == "debug":
+        logger.debug(full_message)
+        pdf_debug_logger.debug(full_message)
+    elif level.lower() == "info":
+        logger.info(full_message)
+        pdf_debug_logger.info(full_message)
+    elif level.lower() == "warning":
+        logger.warning(full_message)
+        pdf_debug_logger.warning(full_message)
+    elif level.lower() == "error":
+        logger.error(full_message)
+        pdf_debug_logger.error(full_message)
+        if error:
+            logger.error(traceback.format_exc())
+            pdf_debug_logger.error(traceback.format_exc())
+    else:
+        logger.info(full_message)
+        pdf_debug_logger.info(full_message)
 
-# Đảm bảo thư mục upload tồn tại
-os.makedirs(TEMP_UPLOAD_DIR, exist_ok=True)
-os.makedirs(STORAGE_DIR, exist_ok=True)
+# Helper function to log debug information during upload
+def log_upload_debug(correlation_id: str, message: str, error: Exception = None):
+    """Log detailed debug information about PDF uploads"""
+    pdf_debug_logger.debug(f"[{correlation_id}] {message}")
+    if error:
+        pdf_debug_logger.error(f"[{correlation_id}] Error: {str(error)}")
+        pdf_debug_logger.error(traceback.format_exc())
 
-# Endpoint upload và xử lý PDF
+# Helper function to send progress updates
+async def send_progress_update(user_id, file_id, step, progress=0.0, message=""):
+    """Send PDF processing progress updates via WebSocket"""
+    try:
+        await send_pdf_upload_progress(user_id, file_id, step, progress, message)
+    except Exception as e:
+        logger.error(f"Error sending progress update: {e}")
+        logger.error(traceback.format_exc())
+
+# Function with fixed indentation for the troublesome parts
+async def handle_pdf_processing_result(result, correlation_id, user_id, file_id, filename, document, vector_status, 
+                                    vector_database_id, temp_file_path, db, is_pdf, mock_mode):
+    """Fixed version of the code with proper indentation"""
+    # If successful, update status but don't try to permanently store files
+    if result.get('success'):
+        try:
+            log_upload_debug(correlation_id, f"Processed file successfully - no permanent storage in Hugging Face environment")
+        except Exception as move_error:
+            log_upload_debug(correlation_id, f"Error in storage handling: {move_error}", move_error)
+        
+        # Update status in PostgreSQL
+        if vector_database_id and document and vector_status:
+            try:
+                log_upload_debug(correlation_id, f"Updating vector status to 'completed' for document ID {document.id}")
+                vector_status.status = "completed"
+                vector_status.embedded_at = datetime.now()
+                vector_status.vector_id = file_id
+                document.is_embedded = True
+                db.commit()
+                log_upload_debug(correlation_id, f"Database status updated successfully")
+            except Exception as db_error:
+                log_upload_debug(correlation_id, f"Error updating database status: {db_error}", db_error)
+        
+        # Send completion notification via WebSocket
+        if user_id:
+            try:
+                await send_pdf_upload_completed(
+                    user_id,
+                    file_id,
+                    filename,
+                    result.get('chunks_processed', 0)
+                )
+                log_upload_debug(correlation_id, f"Sent upload completed notification to user {user_id}")
+            except Exception as ws_error:
+                log_upload_debug(correlation_id, f"Error sending WebSocket notification: {ws_error}", ws_error)
+            
+        # Add document information to the result
+        if document:
+            result["document_database_id"] = document.id
+            
+        # Include mock_mode in response
+        result["mock_mode"] = mock_mode
+    else:
+        log_upload_debug(correlation_id, f"PDF processing failed: {result.get('error', 'Unknown error')}")
+        
+        # Update error status in PostgreSQL
+        if vector_database_id and document and vector_status:
+            try:
+                log_upload_debug(correlation_id, f"Updating vector status to 'failed' for document ID {document.id}")
+                vector_status.status = "failed"
+                vector_status.error_message = result.get('error', 'Unknown error')
+                db.commit()
+                log_upload_debug(correlation_id, f"Database status updated for failure")
+            except Exception as db_error:
+                log_upload_debug(correlation_id, f"Error updating database status for failure: {db_error}", db_error)
+            
+        # Send failure notification via WebSocket
+        if user_id:
+            try:
+                await send_pdf_upload_failed(
+                    user_id,
+                    file_id,
+                    filename,
+                    result.get('error', 'Unknown error')
+                )
+                log_upload_debug(correlation_id, f"Sent upload failed notification to user {user_id}")
+            except Exception as ws_error:
+                log_upload_debug(correlation_id, f"Error sending WebSocket notification: {ws_error}", ws_error)
+        
+    # Cleanup: delete temporary file if it still exists
+    if os.path.exists(temp_file_path):
+        try:
+            os.remove(temp_file_path)
+            log_upload_debug(correlation_id, f"Removed temporary file {temp_file_path}")
+        except Exception as cleanup_error:
+            log_upload_debug(correlation_id, f"Error removing temporary file: {cleanup_error}", cleanup_error)
+    
+    log_upload_debug(correlation_id, f"Upload request completed with success={result.get('success', False)}")
+    return result
+
+# Endpoint for uploading and processing PDFs
 @router.post("/upload", response_model=PDFResponse)
 async def upload_pdf(
     file: UploadFile = File(...),
@@ -40,157 +186,395 @@ async def upload_pdf(
     title: Optional[str] = Form(None),
     description: Optional[str] = Form(None),
     user_id: Optional[str] = Form(None),
-    background_tasks: BackgroundTasks = None
+    vector_database_id: Optional[int] = Form(None),
+    content_type: Optional[str] = Form(None),  # Add content_type parameter
+    background_tasks: BackgroundTasks = None,
+    mock_mode: bool = Form(False),  # Set to False to use real database
+    db: Session = Depends(get_db)
 ):
     """
-    Upload và xử lý file PDF để tạo embeddings và lưu vào Pinecone
+    Upload and process PDF file to create embeddings and store in Pinecone
     
-    - **file**: File PDF cần xử lý
-    - **namespace**: Namespace trong Pinecone để lưu embeddings (mặc định: "Default")
-    - **index_name**: Tên index Pinecone (mặc định: "testbot768")
-    - **title**: Tiêu đề của tài liệu (tùy chọn)
-    - **description**: Mô tả về tài liệu (tùy chọn)
-    - **user_id**: ID của người dùng để cập nhật trạng thái qua WebSocket
+    - **file**: PDF file to process
+    - **namespace**: Namespace in Pinecone to store embeddings (default: "Default")
+    - **index_name**: Name of Pinecone index (default: "testbot768")
+    - **title**: Document title (optional)
+    - **description**: Document description (optional)
+    - **user_id**: User ID for WebSocket status updates
+    - **vector_database_id**: ID of vector database in PostgreSQL (optional)
+    - **content_type**: Content type of the file (optional)
+    - **mock_mode**: Simulate Pinecone operations instead of performing real calls (default: false)
     """
+    # Generate request ID for tracking
+    correlation_id = str(uuid.uuid4())[:8]
+    logger.info(f"[{correlation_id}] PDF upload request received: ns={namespace}, index={index_name}, user={user_id}")
+    log_upload_debug(correlation_id, f"Upload request: vector_db_id={vector_database_id}, mock_mode={mock_mode}")
+    
     try:
-        # Kiểm tra file có phải PDF không
-        if not file.filename.lower().endswith('.pdf'):
-            raise HTTPException(status_code=400, detail="Chỉ chấp nhận file PDF")
+        # Check file type - accept both PDF and plaintext for testing
+        is_pdf = file.filename.lower().endswith('.pdf')
+        is_text = file.filename.lower().endswith(('.txt', '.md', '.html'))
+        
+        log_upload_debug(correlation_id, f"File type check: is_pdf={is_pdf}, is_text={is_text}, filename={file.filename}")
+        
+        if not (is_pdf or is_text):
+            if not mock_mode:
+                # In real mode, only accept PDFs
+                log_upload_debug(correlation_id, f"Rejecting non-PDF file in real mode: {file.filename}")
+                raise HTTPException(status_code=400, detail="Only PDF files are accepted")
+            else:
+                # In mock mode, convert any file to text for testing
+                logger.warning(f"[{correlation_id}] Non-PDF file uploaded in mock mode: {file.filename} - will treat as text")
+        
+        # If vector_database_id provided, get info from PostgreSQL
+        api_key = None
+        vector_db = None
         
-        # Tạo file_id và lưu file tạm
+        if vector_database_id:
+            log_upload_debug(correlation_id, f"Looking up vector database ID {vector_database_id}")
+            
+            vector_db = db.query(VectorDatabase).filter(
+                VectorDatabase.id == vector_database_id,
+                VectorDatabase.status == "active"
+            ).first()
+            if not vector_db:
+                return PDFResponse(
+                    success=False,
+                    error=f"Vector database with ID {vector_database_id} not found or inactive"
+                )
+            
+            log_upload_debug(correlation_id, f"Found vector database: id={vector_db.id}, name={vector_db.name}, index={vector_db.pinecone_index}")
+            
+            # Use vector database information
+            # Try to get API key from relationship
+            log_upload_debug(correlation_id, f"Trying to get API key for vector database {vector_database_id}")
+            
+            # Log available attributes
+            vector_db_attrs = dir(vector_db)
+            log_upload_debug(correlation_id, f"Vector DB attributes: {vector_db_attrs}")
+            
+            if hasattr(vector_db, 'api_key_ref') and vector_db.api_key_ref:
+                log_upload_debug(correlation_id, f"Using API key from relationship for vector database ID {vector_database_id}")
+                log_upload_debug(correlation_id, f"api_key_ref type: {type(vector_db.api_key_ref)}")
+                log_upload_debug(correlation_id, f"api_key_ref attributes: {dir(vector_db.api_key_ref)}")
+                
+                if hasattr(vector_db.api_key_ref, 'key_value'):
+                    api_key = vector_db.api_key_ref.key_value
+                    # Log first few chars of API key for debugging
+                    key_prefix = api_key[:4] + "..." if api_key and len(api_key) > 4 else "invalid/empty"
+                    log_upload_debug(correlation_id, f"API key retrieved: {key_prefix}, length: {len(api_key) if api_key else 0}")
+                    logger.info(f"[{correlation_id}] Using API key from relationship for vector database ID {vector_database_id}")
+                else:
+                    log_upload_debug(correlation_id, f"api_key_ref does not have key_value attribute")
+            elif hasattr(vector_db, 'api_key') and vector_db.api_key:
+                # Fallback to direct api_key if needed (deprecated)
+                api_key = vector_db.api_key
+                key_prefix = api_key[:4] + "..." if api_key and len(api_key) > 4 else "invalid/empty"
+                log_upload_debug(correlation_id, f"Using deprecated direct api_key: {key_prefix}")
+                logger.warning(f"[{correlation_id}] Using deprecated direct api_key for vector database ID {vector_database_id}")
+            else:
+                log_upload_debug(correlation_id, "No API key found in vector database")
+            
+            # Use index from vector database
+            index_name = vector_db.pinecone_index
+            log_upload_debug(correlation_id, f"Using index name '{index_name}' from vector database")
+            logger.info(f"[{correlation_id}] Using index name '{index_name}' from vector database")
+        
+        # Generate file_id and save temporary file
         file_id = str(uuid.uuid4())
-        temp_file_path = os.path.join(TEMP_UPLOAD_DIR, f"{file_id}.pdf")
+        temp_file_path = os.path.join(TEMP_UPLOAD_DIR, f"{file_id}{'.pdf' if is_pdf else '.txt'}")
+        log_upload_debug(correlation_id, f"Generated file_id: {file_id}, temp path: {temp_file_path}")
         
-        # Gửi thông báo bắt đầu xử lý qua WebSocket nếu có user_id
+        # Send notification of upload start via WebSocket if user_id provided
         if user_id:
-            await send_pdf_upload_started(user_id, file.filename, file_id)
+            try:
+                await send_pdf_upload_started(user_id, file.filename, file_id)
+                log_upload_debug(correlation_id, f"Sent upload started notification to user {user_id}")
+            except Exception as ws_error:
+                log_upload_debug(correlation_id, f"Error sending WebSocket notification: {ws_error}", ws_error)
+        
+        # Save file
+        log_upload_debug(correlation_id, f"Reading file content")
+        file_content = await file.read()
+        log_upload_debug(correlation_id, f"File size: {len(file_content)} bytes")
         
-        # Lưu file
         with open(temp_file_path, "wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
+            buffer.write(file_content)
+        log_upload_debug(correlation_id, f"File saved to {temp_file_path}")
             
-        # Tạo metadata
+        # Create metadata
         metadata = {
             "filename": file.filename,
             "content_type": file.content_type
         }
         
+        # Use provided content_type or fallback to file.content_type
+        actual_content_type = content_type or file.content_type
+        log_upload_debug(correlation_id, f"Using content_type: {actual_content_type}")
+        
+        if not actual_content_type:
+            # Fallback content type based on file extension
+            if is_pdf:
+                actual_content_type = "application/pdf"
+            elif is_text:
+                actual_content_type = "text/plain"
+            else:
+                actual_content_type = "application/octet-stream"
+                
+            log_upload_debug(correlation_id, f"No content_type provided, using fallback: {actual_content_type}")
+            
+        metadata["content_type"] = actual_content_type
+        
         if title:
             metadata["title"] = title
+        else:
+            # Use filename as title if not provided
+            title = file.filename
+            metadata["title"] = title
+            
         if description:
             metadata["description"] = description
         
-        # Gửi thông báo tiến độ qua WebSocket
+        # Send progress update via WebSocket
         if user_id:
-            await send_pdf_upload_progress(
+            try:
+                await send_progress_update(
                 user_id, 
                 file_id, 
                 "file_preparation", 
                 0.2, 
                 "File saved, preparing for processing"
             )
+                log_upload_debug(correlation_id, f"Sent file preparation progress to user {user_id}")
+            except Exception as ws_error:
+                log_upload_debug(correlation_id, f"Error sending progress update: {ws_error}", ws_error)
+        
+        # Create document record - do this regardless of mock mode
+        document = None
+        vector_status = None
+        
+        if vector_database_id and vector_db:
+            log_upload_debug(correlation_id, f"Creating PostgreSQL records for document with vector_database_id={vector_database_id}")
             
-        # Khởi tạo PDF processor
-        processor = PDFProcessor(index_name=index_name, namespace=namespace)
+            # Create document record without file content
+            try:
+                document = Document(
+                    name=title or file.filename,
+                    file_type="pdf" if is_pdf else "text",
+                    content_type=actual_content_type,  # Use the actual_content_type here
+                    size=len(file_content),
+                    is_embedded=False,
+                    vector_database_id=vector_database_id
+                )
+                db.add(document)
+                db.commit()
+                db.refresh(document)
+                log_upload_debug(correlation_id, f"Created document record: id={document.id}")
+            except Exception as doc_error:
+                log_upload_debug(correlation_id, f"Error creating document record: {doc_error}", doc_error)
+                raise
+            
+            # Create document content record to store binary data separately
+            try:
+                document_content = DocumentContent(
+                    document_id=document.id,
+                    file_content=file_content
+                )
+                db.add(document_content)
+                db.commit()
+                log_upload_debug(correlation_id, f"Created document content record for document ID {document.id}")
+            except Exception as content_error:
+                log_upload_debug(correlation_id, f"Error creating document content: {content_error}", content_error)
+                raise
+            
+            # Create vector status record
+            try:
+                vector_status = VectorStatus(
+                    document_id=document.id,
+                    vector_database_id=vector_database_id,
+                    status="pending"
+                )
+                db.add(vector_status)
+                db.commit()
+                log_upload_debug(correlation_id, f"Created vector status record for document ID {document.id}")
+            except Exception as status_error:
+                log_upload_debug(correlation_id, f"Error creating vector status: {status_error}", status_error)
+                raise
+            
+            logger.info(f"[{correlation_id}] Created document ID {document.id} and vector status in PostgreSQL")
+            
+        # Initialize PDF processor with correct parameters
+        log_upload_debug(correlation_id, f"Initializing PDFProcessor: index={index_name}, vector_db_id={vector_database_id}, mock_mode={mock_mode}")
+        processor = PDFProcessor(
+            index_name=index_name, 
+            namespace=namespace, 
+            api_key=api_key, 
+            vector_db_id=vector_database_id,
+            mock_mode=mock_mode,
+            correlation_id=correlation_id
+        )
         
-        # Gửi thông báo bắt đầu embedding qua WebSocket
+        # Send embedding start notification via WebSocket
         if user_id:
-            await send_pdf_upload_progress(
+            try:
+                await send_progress_update(
                 user_id, 
                 file_id, 
                 "embedding_start", 
                 0.4, 
                 "Starting to process PDF and create embeddings"
             )
+                log_upload_debug(correlation_id, f"Sent embedding start notification to user {user_id}")
+            except Exception as ws_error:
+                log_upload_debug(correlation_id, f"Error sending WebSocket notification: {ws_error}", ws_error)
         
-        # Xử lý PDF và tạo embeddings
-        # Tạo callback function để xử lý cập nhật tiến độ
-        async def progress_callback_wrapper(step, progress, message):
-            if user_id:
-                await send_progress_update(user_id, file_id, step, progress, message)
-        
-        # Xử lý PDF và tạo embeddings với callback đã được xử lý đúng cách
+        # Process PDF and create embeddings with progress callback
+        log_upload_debug(correlation_id, f"Processing PDF with file_path={temp_file_path}, document_id={file_id}")
         result = await processor.process_pdf(
             file_path=temp_file_path,
             document_id=file_id,
             metadata=metadata,
-            progress_callback=progress_callback_wrapper
+            progress_callback=send_progress_update if user_id else None
         )
         
-        # Nếu thành công, chuyển file vào storage
-        if result.get('success'):
-            storage_path = os.path.join(STORAGE_DIR, f"{file_id}.pdf")
-            shutil.move(temp_file_path, storage_path)
-            
-            # Gửi thông báo hoàn thành qua WebSocket
-            if user_id:
-                await send_pdf_upload_completed(
-                    user_id,
-                    file_id,
-                    file.filename,
-                    result.get('chunks_processed', 0)
-                )
-        else:
-            # Gửi thông báo lỗi qua WebSocket
-            if user_id:
-                await send_pdf_upload_failed(
-                    user_id,
-                    file_id,
-                    file.filename,
-                    result.get('error', 'Unknown error')
-                )
-            
-        # Dọn dẹp: xóa file tạm nếu vẫn còn
-        if os.path.exists(temp_file_path):
-            os.remove(temp_file_path)
-            
-        return result
+        log_upload_debug(correlation_id, f"PDF processing result: {result}")
+        
+        # Handle PDF processing result
+        return await handle_pdf_processing_result(result, correlation_id, user_id, file_id, file.filename, document, vector_status, 
+                                                vector_database_id, temp_file_path, db, is_pdf, mock_mode)
     except Exception as e:
-        # Dọn dẹp nếu có lỗi
-        if 'temp_file_path' in locals() and os.path.exists(temp_file_path):
+        return await handle_upload_error(e, correlation_id, temp_file_path, user_id, file_id, file.filename, vector_database_id, vector_status, db, mock_mode)
+
+# Error handling for upload_pdf function
+async def handle_upload_error(e, correlation_id, temp_file_path, user_id, file_id, filename, vector_database_id, vector_status, db, mock_mode):
+    """Fixed version of the error handling part with proper indentation"""
+    log_upload_debug(correlation_id, f"Error in upload_pdf: {str(e)}", e)
+    logger.exception(f"[{correlation_id}] Error in upload_pdf: {str(e)}")
+    
+    # Cleanup on error
+    if os.path.exists(temp_file_path):
+        try:
             os.remove(temp_file_path)
-            
-        # Gửi thông báo lỗi qua WebSocket
-        if 'user_id' in locals() and user_id and 'file_id' in locals():
+            log_upload_debug(correlation_id, f"Cleaned up temp file after error: {temp_file_path}")
+        except Exception as cleanup_error:
+            log_upload_debug(correlation_id, f"Error cleaning up temporary file: {cleanup_error}", cleanup_error)
+        
+    # Update error status in PostgreSQL
+    if vector_database_id and vector_status:
+        try:
+            vector_status.status = "failed"
+            vector_status.error_message = str(e)
+            db.commit()
+            log_upload_debug(correlation_id, f"Updated database with error status")
+        except Exception as db_error:
+            log_upload_debug(correlation_id, f"Error updating database with error status: {db_error}", db_error)
+        
+    # Send failure notification via WebSocket
+    if user_id and file_id:
+        try:
             await send_pdf_upload_failed(
                 user_id,
                 file_id,
-                file.filename,
+                filename,
                 str(e)
             )
+            log_upload_debug(correlation_id, f"Sent failure notification for exception")
+        except Exception as ws_error:
+            log_upload_debug(correlation_id, f"Error sending WebSocket notification for failure: {ws_error}", ws_error)
             
-        return PDFResponse(
-            success=False,
-            error=str(e)
-        )
-
-# Function để gửi cập nhật tiến độ - được sử dụng trong callback
-async def send_progress_update(user_id, document_id, step, progress, message):
-    if user_id:
-        await send_pdf_upload_progress(user_id, document_id, step, progress, message)
+    log_upload_debug(correlation_id, f"Upload request failed with exception: {str(e)}")
+    return PDFResponse(
+        success=False,
+        error=str(e),
+        mock_mode=mock_mode
+    )
 
 # Endpoint xóa tài liệu
 @router.delete("/namespace", response_model=PDFResponse)
 async def delete_namespace(
     namespace: str = "Default",
     index_name: str = "testbot768",
-    user_id: Optional[str] = None
+    vector_database_id: Optional[int] = None,
+    user_id: Optional[str] = None,
+    db: Session = Depends(get_db)
 ):
     """
     Xóa toàn bộ embeddings trong một namespace từ Pinecone (tương ứng xoá namespace)
 
     - **namespace**: Namespace trong Pinecone (mặc định: "Default")
     - **index_name**: Tên index Pinecone (mặc định: "testbot768")
+    - **vector_database_id**: ID của vector database trong PostgreSQL (nếu có)
     - **user_id**: ID của người dùng để cập nhật trạng thái qua WebSocket
     """
+    logger.info(f"Delete namespace request: namespace={namespace}, index={index_name}, vector_db_id={vector_database_id}")
+    
     try:
+        # Nếu có vector_database_id, lấy thông tin từ PostgreSQL
+        api_key = None
+        vector_db = None
+        mock_mode = False  # Use real mode by default
+
+        if vector_database_id:
+            vector_db = db.query(VectorDatabase).filter(
+                VectorDatabase.id == vector_database_id,
+                VectorDatabase.status == "active"
+            ).first()
+            if not vector_db:
+                return PDFResponse(
+                    success=False,
+                    error=f"Vector database with ID {vector_database_id} not found or inactive"
+                )
+            
+            # Use index from vector database
+            index_name = vector_db.pinecone_index
+            
+            # Get API key
+            if hasattr(vector_db, 'api_key_ref') and vector_db.api_key_ref:
+                api_key = vector_db.api_key_ref.key_value
+            elif hasattr(vector_db, 'api_key') and vector_db.api_key:
+                api_key = vector_db.api_key
+            
+            # Use namespace based on vector database ID
+            namespace = f"vdb-{vector_database_id}" if vector_database_id else namespace
+            logger.info(f"Using namespace '{namespace}' based on vector database ID")
+            
         # Gửi thông báo bắt đầu xóa qua WebSocket
         if user_id:
             await send_pdf_delete_started(user_id, namespace)
             
-        processor = PDFProcessor(index_name=index_name, namespace=namespace)
+        processor = PDFProcessor(
+            index_name=index_name, 
+            namespace=namespace,
+            api_key=api_key,
+            vector_db_id=vector_database_id,
+            mock_mode=mock_mode
+        )
         result = await processor.delete_namespace()
         
+        # If in mock mode, also update PostgreSQL to reflect the deletion
+        if mock_mode and result.get('success') and vector_database_id:
+            try:
+                # Update vector statuses for this database
+                affected_count = db.query(VectorStatus).filter(
+                    VectorStatus.vector_database_id == vector_database_id,
+                    VectorStatus.status != "deleted"
+                ).update({"status": "deleted", "updated_at": datetime.now()})
+                
+                # Update document embedding status
+                db.query(Document).filter(
+                    Document.vector_database_id == vector_database_id,
+                    Document.is_embedded == True
+                ).update({"is_embedded": False})
+                
+                db.commit()
+                logger.info(f"Updated {affected_count} vector statuses to 'deleted'")
+                
+                # Include this info in the result
+                result["updated_records"] = affected_count
+            except Exception as db_error:
+                logger.error(f"Error updating PostgreSQL records after namespace deletion: {db_error}")
+                result["postgresql_update_error"] = str(db_error)
+        
         # Gửi thông báo kết quả qua WebSocket
         if user_id:
             if result.get('success'):
@@ -200,6 +584,8 @@ async def delete_namespace(
                 
         return result
     except Exception as e:
+        logger.exception(f"Error in delete_namespace: {str(e)}")
+        
         # Gửi thông báo lỗi qua WebSocket
         if user_id:
             await send_pdf_delete_failed(user_id, namespace, str(e))
@@ -211,23 +597,235 @@ async def delete_namespace(
 
 # Endpoint lấy danh sách tài liệu
 @router.get("/documents", response_model=DocumentsListResponse)
-async def get_documents(namespace: str = "Default", index_name: str = "testbot768"):
+async def get_documents(
+    namespace: str = "Default", 
+    index_name: str = "testbot768",
+    vector_database_id: Optional[int] = None,
+    db: Session = Depends(get_db)
+):
     """
     Lấy thông tin về tất cả tài liệu đã được embed
     
     - **namespace**: Namespace trong Pinecone (mặc định: "Default")
     - **index_name**: Tên index Pinecone (mặc định: "testbot768")
+    - **vector_database_id**: ID của vector database trong PostgreSQL (nếu có)
     """
+    logger.info(f"Get documents request: namespace={namespace}, index={index_name}, vector_db_id={vector_database_id}")
+    
     try:
+        # Nếu có vector_database_id, lấy thông tin từ PostgreSQL
+        api_key = None
+        vector_db = None
+        mock_mode = False  # Use real mode by default
+
+        if vector_database_id:
+            vector_db = db.query(VectorDatabase).filter(
+                VectorDatabase.id == vector_database_id,
+                VectorDatabase.status == "active"
+            ).first()
+            
+            if not vector_db:
+                return DocumentsListResponse(
+                    success=False,
+                    error=f"Vector database with ID {vector_database_id} not found or inactive"
+                )
+                
+            # Use index from vector database
+            index_name = vector_db.pinecone_index
+            
+            # Get API key
+            if hasattr(vector_db, 'api_key_ref') and vector_db.api_key_ref:
+                api_key = vector_db.api_key_ref.key_value
+            elif hasattr(vector_db, 'api_key') and vector_db.api_key:
+                api_key = vector_db.api_key
+                
+            # Use namespace based on vector database ID
+            namespace = f"vdb-{vector_database_id}" if vector_database_id else namespace
+            logger.info(f"Using namespace '{namespace}' based on vector database ID")
+            
         # Khởi tạo PDF processor
-        processor = PDFProcessor(index_name=index_name, namespace=namespace)
+        processor = PDFProcessor(
+            index_name=index_name, 
+            namespace=namespace,
+            api_key=api_key,
+            vector_db_id=vector_database_id,
+            mock_mode=mock_mode
+        )
         
-        # Lấy danh sách documents
-        result = await processor.list_documents()
+        # Lấy danh sách documents từ Pinecone
+        pinecone_result = await processor.list_documents()
         
-        return result
+        # If vector_database_id is provided, also fetch from PostgreSQL
+        if vector_database_id:
+            try:
+                # Get all successfully embedded documents for this vector database
+                documents = db.query(Document).join(
+                    VectorStatus, Document.id == VectorStatus.document_id
+                ).filter(
+                    Document.vector_database_id == vector_database_id,
+                    Document.is_embedded == True,
+                    VectorStatus.status == "completed"
+                ).all()
+                
+                # Add document info to the result
+                if documents:
+                    pinecone_result["postgresql_documents"] = [
+                        {
+                            "id": doc.id,
+                            "name": doc.name,
+                            "file_type": doc.file_type,
+                            "content_type": doc.content_type,
+                            "created_at": doc.created_at.isoformat() if doc.created_at else None
+                        }
+                        for doc in documents
+                    ]
+                    pinecone_result["postgresql_document_count"] = len(documents)
+            except Exception as db_error:
+                logger.error(f"Error fetching PostgreSQL documents: {db_error}")
+                pinecone_result["postgresql_error"] = str(db_error)
+        
+        return pinecone_result
     except Exception as e:
+        logger.exception(f"Error in get_documents: {str(e)}")
+        
         return DocumentsListResponse(
             success=False,
             error=str(e)
-        ) 
+        ) 
+
+# Health check endpoint for PDF API
+@router.get("/health")
+async def health_check():
+    return {
+        "status": "healthy",
+        "version": "1.0.0",
+        "message": "PDF API is running"
+    }
+
+# Document deletion endpoint
+@router.delete("/document", response_model=PDFResponse)
+async def delete_document(
+    document_id: str,
+    namespace: str = "Default",
+    index_name: str = "testbot768",
+    vector_database_id: Optional[int] = None,
+    user_id: Optional[str] = None,
+    mock_mode: bool = False,
+    db: Session = Depends(get_db)
+):
+    """
+    Delete vectors for a specific document from the vector database
+    
+    - **document_id**: ID of the document to delete
+    - **namespace**: Namespace in the vector database (default: "Default")
+    - **index_name**: Name of the vector index (default: "testbot768")
+    - **vector_database_id**: ID of vector database in PostgreSQL (optional)
+    - **user_id**: User ID for WebSocket status updates (optional)
+    - **mock_mode**: Simulate vector database operations (default: false)
+    """
+    logger.info(f"Delete document request: document_id={document_id}, namespace={namespace}, index={index_name}, vector_db_id={vector_database_id}, mock_mode={mock_mode}")
+    
+    try:
+        # If vector_database_id is provided, get info from PostgreSQL
+        api_key = None
+        vector_db = None
+
+        if vector_database_id:
+            vector_db = db.query(VectorDatabase).filter(
+                VectorDatabase.id == vector_database_id,
+                VectorDatabase.status == "active"
+            ).first()
+            if not vector_db:
+                return PDFResponse(
+                    success=False,
+                    error=f"Vector database with ID {vector_database_id} not found or inactive"
+                )
+            
+            # Use index from vector database
+            index_name = vector_db.pinecone_index
+            
+            # Get API key
+            if hasattr(vector_db, 'api_key_ref') and vector_db.api_key_ref:
+                api_key = vector_db.api_key_ref.key_value
+            elif hasattr(vector_db, 'api_key') and vector_db.api_key:
+                api_key = vector_db.api_key
+            
+            # Use namespace based on vector database ID
+            namespace = f"vdb-{vector_database_id}" if vector_database_id else namespace
+            logger.info(f"Using namespace '{namespace}' based on vector database ID")
+            
+        # Send notification of deletion start via WebSocket if user_id provided
+        if user_id:
+            try:
+                await send_pdf_delete_started(user_id, document_id)
+            except Exception as ws_error:
+                logger.error(f"Error sending WebSocket notification: {ws_error}")
+        
+        # Initialize PDF processor
+        processor = PDFProcessor(
+            index_name=index_name, 
+            namespace=namespace, 
+            api_key=api_key, 
+            vector_db_id=vector_database_id,
+            mock_mode=mock_mode
+        )
+        
+        # Delete document vectors
+        result = await processor.delete_document(document_id)
+        
+        # If successful and vector_database_id is provided, update PostgreSQL records
+        if result.get('success') and vector_database_id:
+            try:
+                # Find document by vector ID if it exists
+                document = db.query(Document).join(
+                    VectorStatus, Document.id == VectorStatus.document_id
+                ).filter(
+                    Document.vector_database_id == vector_database_id,
+                    VectorStatus.vector_id == document_id
+                ).first()
+                
+                if document:
+                    # Update vector status
+                    vector_status = db.query(VectorStatus).filter(
+                        VectorStatus.document_id == document.id,
+                        VectorStatus.vector_database_id == vector_database_id
+                    ).first()
+                    
+                    if vector_status:
+                        vector_status.status = "deleted"
+                        db.commit()
+                        result["postgresql_updated"] = True
+                        logger.info(f"Updated vector status for document ID {document.id} to 'deleted'")
+            except Exception as db_error:
+                logger.error(f"Error updating PostgreSQL records: {db_error}")
+                result["postgresql_error"] = str(db_error)
+        
+        # Send notification of deletion completion via WebSocket if user_id provided
+        if user_id:
+            try:
+                if result.get('success'):
+                    await send_pdf_delete_completed(user_id, document_id)
+                else:
+                    await send_pdf_delete_failed(user_id, document_id, result.get('error', 'Unknown error'))
+            except Exception as ws_error:
+                logger.error(f"Error sending WebSocket notification: {ws_error}")
+        
+        return result
+    except Exception as e:
+        logger.exception(f"Error in delete_document: {str(e)}")
+        
+        # Send notification of deletion failure via WebSocket if user_id provided
+        if user_id:
+            try:
+                await send_pdf_delete_failed(user_id, document_id, str(e))
+            except Exception as ws_error:
+                logger.error(f"Error sending WebSocket notification: {ws_error}")
+        
+        return PDFResponse(
+            success=False,
+            error=str(e),
+            mock_mode=mock_mode
+        )
+
+
+

app/api/pdf_websocket.py

@@ -108,7 +108,184 @@ class ConnectionManager:
 # Tạo instance của ConnectionManager
 manager = ConnectionManager()
 
-@router.websocket("/pdf/{user_id}")
+# Test route for manual WebSocket sending
+@router.get("/ws/test/{user_id}")
+async def test_websocket_send(user_id: str):
+    """
+    Test route to manually send a WebSocket message to a user
+    This is useful for debugging WebSocket connections
+    """
+    logger.info(f"Attempting to send test message to user: {user_id}")
+    
+    # Check if user has a connection
+    status = manager.get_connection_status(user_id)
+    if not status["active"]:
+        logger.warning(f"No active WebSocket connection for user: {user_id}")
+        return {"success": False, "message": f"No active WebSocket connection for user: {user_id}"}
+    
+    # Send test message
+    await manager.send_message({
+        "type": "test_message",
+        "message": "This is a test WebSocket message",
+        "timestamp": int(time.time())
+    }, user_id)
+    
+    logger.info(f"Test message sent to user: {user_id}")
+    return {"success": True, "message": f"Test message sent to user: {user_id}"}
+
+@router.websocket("/ws/pdf/{user_id}")
+async def websocket_endpoint(websocket: WebSocket, user_id: str):
+    """Endpoint WebSocket để cập nhật tiến trình xử lý PDF"""
+    logger.info(f"WebSocket connection request received for user: {user_id}")
+    
+    try:
+        await manager.connect(websocket, user_id)
+        logger.info(f"WebSocket connection accepted for user: {user_id}")
+        
+        # Send a test message to confirm connection
+        await manager.send_message({
+            "type": "connection_established",
+            "message": "WebSocket connection established successfully",
+            "user_id": user_id,
+            "timestamp": int(time.time())
+        }, user_id)
+        
+        try:
+            while True:
+                # Đợi tin nhắn từ client (chỉ để giữ kết nối)
+                data = await websocket.receive_text()
+                logger.debug(f"Received from client: {data}")
+                
+                # Echo back to confirm receipt
+                if data != "heartbeat":  # Don't echo heartbeats
+                    await manager.send_message({
+                        "type": "echo",
+                        "message": f"Received: {data}",
+                        "timestamp": int(time.time())
+                    }, user_id)
+        except WebSocketDisconnect:
+            logger.info(f"WebSocket disconnected for user: {user_id}")
+            manager.disconnect(websocket, user_id)
+        except Exception as e:
+            logger.error(f"WebSocket error: {str(e)}")
+            manager.disconnect(websocket, user_id)
+    except Exception as e:
+        logger.error(f"Failed to establish WebSocket connection: {str(e)}")
+        # Ensure the connection is closed properly
+        if websocket.client_state != 4:  # 4 = CLOSED
+            await websocket.close(code=1011, reason=f"Server error: {str(e)}")
+
+import logging
+from typing import Dict, List, Optional, Any
+from fastapi import WebSocket, WebSocketDisconnect, APIRouter
+from pydantic import BaseModel
+import json
+import time
+
+# Cấu hình logging
+logger = logging.getLogger(__name__)
+
+# Models cho Swagger documentation
+class ConnectionStatus(BaseModel):
+    user_id: str
+    active: bool
+    connection_count: int
+    last_activity: Optional[float] = None
+
+class UserConnection(BaseModel):
+    user_id: str
+    connection_count: int
+
+class AllConnectionsStatus(BaseModel):
+    total_users: int
+    total_connections: int
+    users: List[UserConnection]
+
+# Khởi tạo router
+router = APIRouter(
+    prefix="",
+    tags=["WebSockets"],
+)
+
+class ConnectionManager:
+    """Quản lý các kết nối WebSocket"""
+    
+    def __init__(self):
+        # Lưu trữ các kết nối theo user_id
+        self.active_connections: Dict[str, List[WebSocket]] = {}
+        
+    async def connect(self, websocket: WebSocket, user_id: str):
+        """Kết nối một WebSocket mới"""
+        await websocket.accept()
+        if user_id not in self.active_connections:
+            self.active_connections[user_id] = []
+        self.active_connections[user_id].append(websocket)
+        logger.info(f"New WebSocket connection for user {user_id}. Total connections: {len(self.active_connections[user_id])}")
+        
+    def disconnect(self, websocket: WebSocket, user_id: str):
+        """Ngắt kết nối WebSocket"""
+        if user_id in self.active_connections:
+            if websocket in self.active_connections[user_id]:
+                self.active_connections[user_id].remove(websocket)
+            # Xóa user_id khỏi dict nếu không còn kết nối nào
+            if not self.active_connections[user_id]:
+                del self.active_connections[user_id]
+        logger.info(f"WebSocket disconnected for user {user_id}")
+        
+    async def send_message(self, message: Dict[str, Any], user_id: str):
+        """Gửi tin nhắn tới tất cả kết nối của một user"""
+        if user_id in self.active_connections:
+            disconnected_websockets = []
+            for websocket in self.active_connections[user_id]:
+                try:
+                    await websocket.send_text(json.dumps(message))
+                except Exception as e:
+                    logger.error(f"Error sending message to WebSocket: {str(e)}")
+                    disconnected_websockets.append(websocket)
+            
+            # Xóa các kết nối bị ngắt
+            for websocket in disconnected_websockets:
+                self.disconnect(websocket, user_id)
+    
+    def get_connection_status(self, user_id: str = None) -> Dict[str, Any]:
+        """Lấy thông tin về trạng thái kết nối WebSocket"""
+        if user_id:
+            # Trả về thông tin kết nối cho user cụ thể
+            if user_id in self.active_connections:
+                return {
+                    "user_id": user_id,
+                    "active": True,
+                    "connection_count": len(self.active_connections[user_id]),
+                    "last_activity": time.time()
+                }
+            else:
+                return {
+                    "user_id": user_id,
+                    "active": False,
+                    "connection_count": 0,
+                    "last_activity": None
+                }
+        else:
+            # Trả về thông tin tất cả kết nối
+            result = {
+                "total_users": len(self.active_connections),
+                "total_connections": sum(len(connections) for connections in self.active_connections.values()),
+                "users": []
+            }
+            
+            for uid, connections in self.active_connections.items():
+                result["users"].append({
+                    "user_id": uid,
+                    "connection_count": len(connections)
+                })
+            
+            return result
+
+
+# Tạo instance của ConnectionManager
+manager = ConnectionManager()
+
+@router.websocket("/ws/pdf/{user_id}")
 async def websocket_endpoint(websocket: WebSocket, user_id: str):
     """Endpoint WebSocket để cập nhật tiến trình xử lý PDF"""
     await manager.connect(websocket, user_id)
@@ -123,7 +300,7 @@ async def websocket_endpoint(websocket: WebSocket, user_id: str):
         manager.disconnect(websocket, user_id)
 
 # API endpoints để kiểm tra trạng thái WebSocket
-@router.get("/status", response_model=AllConnectionsStatus, responses={
+@router.get("/ws/status", response_model=AllConnectionsStatus, responses={
     200: {
         "description": "Successful response",
         "content": {
@@ -151,7 +328,7 @@ async def get_all_websocket_connections():
     """
     return manager.get_connection_status()
 
-@router.get("/status/{user_id}", response_model=ConnectionStatus, responses={
+@router.get("/ws/status/{user_id}", response_model=ConnectionStatus, responses={
     200: {
         "description": "Successful response for active connection",
         "content": {
@@ -245,11 +422,12 @@ async def send_pdf_delete_started(user_id: str, namespace: str):
         "timestamp": int(time.time())
     }, user_id)
 
-async def send_pdf_delete_completed(user_id: str, namespace: str):
+async def send_pdf_delete_completed(user_id: str, namespace: str, deleted_count: int = 0):
     """Gửi thông báo hoàn thành xóa PDF"""
     await manager.send_message({
         "type": "pdf_delete_completed",
         "namespace": namespace,
+        "deleted_count": deleted_count,
         "timestamp": int(time.time())
     }, user_id)
 

app/api/postgresql_routes.py

@@ -4,8 +4,10 @@ import traceback
 from datetime import datetime, timedelta, timezone
 import time
 from functools import lru_cache
+from pathlib import Path as pathlib_Path  # Import Path from pathlib with a different name
 
-from fastapi import APIRouter, HTTPException, Depends, Query, Path, Body
+from fastapi import APIRouter, HTTPException, Depends, Query, Body, Response, File, UploadFile, Form, BackgroundTasks
+from fastapi.params import Path  # Import Path explicitly from fastapi.params instead
 from sqlalchemy.orm import Session
 from sqlalchemy.exc import SQLAlchemyError
 from typing import List, Optional, Dict, Any
@@ -16,9 +18,10 @@ from sqlalchemy import text, inspect, func
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy import desc, func
 from cachetools import TTLCache
+import uuid
 
 from app.database.postgresql import get_db
-from app.database.models import FAQItem, EmergencyItem, EventItem, AboutPixity, SolanaSummit, DaNangBucketList, ApiKey, VectorDatabase, Document, VectorStatus, TelegramBot, ChatEngine, BotEngine, EngineVectorDb
+from app.database.models import FAQItem, EmergencyItem, EventItem, AboutPixity, SolanaSummit, DaNangBucketList, ApiKey, VectorDatabase, Document, VectorStatus, TelegramBot, ChatEngine, BotEngine, EngineVectorDb, DocumentContent
 from pydantic import BaseModel, Field, ConfigDict
 
 # Configure logging
@@ -1713,7 +1716,8 @@ async def update_solana_summit(
 
 # --- API Key models and endpoints ---
 class ApiKeyBase(BaseModel):
-    name: str
+    key_type: str
+    key_value: str
     description: Optional[str] = None
     is_active: bool = True
 
@@ -1721,13 +1725,13 @@ class ApiKeyCreate(ApiKeyBase):
     pass
 
 class ApiKeyUpdate(BaseModel):
-    name: Optional[str] = None
+    key_type: Optional[str] = None
+    key_value: Optional[str] = None
     description: Optional[str] = None
     is_active: Optional[bool] = None
 
 class ApiKeyResponse(ApiKeyBase):
     id: int
-    key: str
     created_at: datetime
     last_used: Optional[datetime] = None
     
@@ -1772,23 +1776,10 @@ async def create_api_key(
     Create a new API key.
     """
     try:
-        # Generate a secure API key
-        import secrets
-        import string
-        import time
-        
-        # Create a random key with a prefix for easier identification
-        prefix = "px_"
-        random_key = ''.join(secrets.choice(string.ascii_letters + string.digits) for _ in range(32))
-        timestamp = hex(int(time.time()))[2:]
-        
-        # Combine parts for the final key
-        key_value = f"{prefix}{timestamp}_{random_key}"
-        
         # Create API key object
         db_api_key = ApiKey(
-            name=api_key.name,
-            key=key_value,
+            key_type=api_key.key_type,
+            key_value=api_key.key_value,
             description=api_key.description,
             is_active=api_key.is_active
         )
@@ -1844,8 +1835,10 @@ async def update_api_key(
             raise HTTPException(status_code=404, detail=f"API key with ID {api_key_id} not found")
         
         # Update fields if provided
-        if api_key_update.name is not None:
-            db_api_key.name = api_key_update.name
+        if api_key_update.key_type is not None:
+            db_api_key.key_type = api_key_update.key_type
+        if api_key_update.key_value is not None:
+            db_api_key.key_value = api_key_update.key_value
         if api_key_update.description is not None:
             db_api_key.description = api_key_update.description
         if api_key_update.is_active is not None:
@@ -1905,17 +1898,17 @@ async def validate_api_key(
     Validate an API key and update its last_used timestamp.
     """
     try:
-        db_api_key = db.query(ApiKey).filter(ApiKey.key == key, ApiKey.is_active == True).first()
+        db_api_key = db.query(ApiKey).filter(ApiKey.key_value == key, ApiKey.is_active == True).first()
         if not db_api_key:
             return {"valid": False, "message": "Invalid or inactive API key"}
-        
-        # Update last_used timestamp
-        db_api_key.last_used = datetime.utcnow()
+            
+        # Update last used timestamp
+        db_api_key.last_used = datetime.now()
         db.commit()
         
         return {
             "valid": True,
-            "name": db_api_key.name,
+            "key_type": db_api_key.key_type,
             "id": db_api_key.id,
             "message": "API key is valid"
         }
@@ -1929,23 +1922,30 @@ class VectorDatabaseBase(BaseModel):
     name: str
     description: Optional[str] = None
     pinecone_index: str
-    api_key: str
+    api_key_id: Optional[int] = None  # Make api_key_id optional to handle NULL values
     status: str = "active"
 
 class VectorDatabaseCreate(VectorDatabaseBase):
+    api_key_id: int  # Keep this required for new databases
     pass
 
 class VectorDatabaseUpdate(BaseModel):
     name: Optional[str] = None
     description: Optional[str] = None
     pinecone_index: Optional[str] = None
-    api_key: Optional[str] = None
+    api_key_id: Optional[int] = None
     status: Optional[str] = None
 
-class VectorDatabaseResponse(VectorDatabaseBase):
+class VectorDatabaseResponse(BaseModel):
+    name: str
+    description: Optional[str] = None
+    pinecone_index: str
+    api_key_id: Optional[int] = None  # Make api_key_id optional to handle NULL values
+    status: str
     id: int
     created_at: datetime
     updated_at: datetime
+    message: Optional[str] = None  # Add message field for notifications
     
     model_config = ConfigDict(from_attributes=True)
 
@@ -1960,6 +1960,7 @@ class VectorDatabaseDetailResponse(BaseModel):
     document_count: int
     embedded_count: int
     pending_count: int
+    message: Optional[str] = None  # Add message field for notifications
     
     model_config = ConfigDict(from_attributes=True)
 
@@ -1999,7 +2000,7 @@ async def create_vector_database(
     db: Session = Depends(get_db)
 ):
     """
-    Create a new vector database.
+    Create a new vector database. If the specified Pinecone index doesn't exist, it will be created automatically.
     """
     try:
         # Check if a database with the same name already exists
@@ -2007,6 +2008,71 @@ async def create_vector_database(
         if existing_db:
             raise HTTPException(status_code=400, detail=f"Vector database with name '{vector_db.name}' already exists")
         
+        # Check if the API key exists
+        api_key = db.query(ApiKey).filter(ApiKey.id == vector_db.api_key_id).first()
+        if not api_key:
+            raise HTTPException(status_code=400, detail=f"API key with ID {vector_db.api_key_id} not found")
+        
+        # Initialize Pinecone client with the API key
+        from pinecone import Pinecone, ServerlessSpec
+        pc_client = Pinecone(api_key=api_key.key_value)
+        
+        # Check if the index exists
+        index_list = pc_client.list_indexes()
+        index_names = index_list.names() if hasattr(index_list, 'names') else []
+        
+        index_exists = vector_db.pinecone_index in index_names
+        index_created = False
+        
+        if not index_exists:
+            # Index doesn't exist - try to create it
+            try:
+                logger.info(f"Pinecone index '{vector_db.pinecone_index}' does not exist. Attempting to create it automatically.")
+                
+                # Create the index with standard parameters
+                pc_client.create_index(
+                    name=vector_db.pinecone_index,
+                    dimension=1536,  # Standard OpenAI embedding dimension
+                    metric="cosine",  # Most common similarity metric
+                    spec=ServerlessSpec(
+                        cloud="aws",
+                        region="us-east-1"  # Use a standard region that works with the free tier
+                    )
+                )
+                
+                logger.info(f"Successfully created Pinecone index '{vector_db.pinecone_index}'")
+                index_created = True
+                
+                # Allow some time for the index to initialize
+                import time
+                time.sleep(5)
+                
+            except Exception as create_error:
+                logger.error(f"Failed to create Pinecone index '{vector_db.pinecone_index}': {create_error}")
+                raise HTTPException(
+                    status_code=400, 
+                    detail=f"Failed to create Pinecone index '{vector_db.pinecone_index}': {str(create_error)}"
+                )
+        
+        # Verify we can connect to the index (whether existing or newly created)
+        try:
+            index = pc_client.Index(vector_db.pinecone_index)
+            # Try to get stats to verify connection
+            stats = index.describe_index_stats()
+            
+            # Create success message based on whether we created the index or used an existing one
+            if index_created:
+                success_message = f"Successfully created and connected to new Pinecone index '{vector_db.pinecone_index}'"
+            else:
+                success_message = f"Successfully connected to existing Pinecone index '{vector_db.pinecone_index}'"
+                
+            logger.info(f"{success_message}: {stats}")
+            
+        except Exception as e:
+            error_message = f"Error connecting to Pinecone index '{vector_db.pinecone_index}': {str(e)}"
+            logger.error(error_message)
+            raise HTTPException(status_code=400, detail=error_message)
+        
         # Create new vector database
         db_vector_db = VectorDatabase(**vector_db.model_dump())
         
@@ -2014,7 +2080,16 @@ async def create_vector_database(
         db.commit()
         db.refresh(db_vector_db)
         
-        return VectorDatabaseResponse.model_validate(db_vector_db, from_attributes=True)
+        # Return response with additional info about index creation
+        response_data = VectorDatabaseResponse.model_validate(db_vector_db, from_attributes=True).model_dump()
+        
+        # Add a message to the response indicating whether the index was created or existed
+        if index_created:
+            response_data["message"] = f"Created new Pinecone index '{vector_db.pinecone_index}' automatically"
+        else:
+            response_data["message"] = f"Using existing Pinecone index '{vector_db.pinecone_index}'"
+            
+        return VectorDatabaseResponse.model_validate(response_data)
     except HTTPException:
         raise
     except SQLAlchemyError as e:
@@ -2068,6 +2143,12 @@ async def update_vector_database(
             if existing_db:
                 raise HTTPException(status_code=400, detail=f"Vector database with name '{vector_db_update.name}' already exists")
         
+        # Check if API key exists if updating API key ID
+        if vector_db_update.api_key_id:
+            api_key = db.query(ApiKey).filter(ApiKey.id == vector_db_update.api_key_id).first()
+            if not api_key:
+                raise HTTPException(status_code=400, detail=f"API key with ID {vector_db_update.api_key_id} not found")
+        
         # Update fields if provided
         update_data = vector_db_update.model_dump(exclude_unset=True)
         for key, value in update_data.items():
@@ -2149,6 +2230,7 @@ async def get_vector_database_info(
 ):
     """
     Get detailed information about a vector database including document counts.
+    Also verifies connectivity to the Pinecone index.
     """
     try:
         # Get the vector database
@@ -2173,6 +2255,40 @@ async def get_vector_database_info(
             Document.is_embedded == False
         ).scalar()
         
+        # Verify Pinecone index connectivity if API key is available
+        message = None
+        if vector_db.api_key_id:
+            try:
+                # Get the API key
+                api_key = db.query(ApiKey).filter(ApiKey.id == vector_db.api_key_id).first()
+                if api_key:
+                    # Initialize Pinecone client with the API key
+                    from pinecone import Pinecone
+                    pc_client = Pinecone(api_key=api_key.key_value)
+                    
+                    # Check if the index exists
+                    index_list = pc_client.list_indexes()
+                    index_names = index_list.names() if hasattr(index_list, 'names') else []
+                    
+                    if vector_db.pinecone_index in index_names:
+                        # Try to connect to the index
+                        index = pc_client.Index(vector_db.pinecone_index)
+                        stats = index.describe_index_stats()
+                        message = f"Pinecone index '{vector_db.pinecone_index}' is operational with {stats.get('total_vector_count', 0)} vectors"
+                        logger.info(f"Successfully connected to Pinecone index '{vector_db.pinecone_index}': {stats}")
+                    else:
+                        message = f"Pinecone index '{vector_db.pinecone_index}' does not exist. Available indexes: {', '.join(index_names)}"
+                        logger.warning(message)
+                else:
+                    message = f"API key with ID {vector_db.api_key_id} not found"
+                    logger.warning(message)
+            except Exception as e:
+                message = f"Error connecting to Pinecone: {str(e)}"
+                logger.error(message)
+        else:
+            message = "No API key associated with this vector database"
+            logger.warning(message)
+        
         # Create response with added counts
         result = VectorDatabaseDetailResponse(
             id=vector_db.id,
@@ -2184,7 +2300,8 @@ async def get_vector_database_info(
             updated_at=vector_db.updated_at,
             document_count=total_docs or 0,
             embedded_count=embedded_docs or 0,
-            pending_count=pending_docs or 0
+            pending_count=pending_docs or 0,
+            message=message
         )
         
         return result
@@ -2198,29 +2315,26 @@ async def get_vector_database_info(
 # --- Document models and endpoints ---
 class DocumentBase(BaseModel):
     name: str
-    content_type: str
     vector_database_id: int
-    file_metadata: Optional[Dict[str, Any]] = None
 
-class DocumentCreate(DocumentBase):
-    pass
+class DocumentCreate(BaseModel):
+    name: str
+    vector_database_id: int
 
 class DocumentUpdate(BaseModel):
     name: Optional[str] = None
-    file_metadata: Optional[Dict[str, Any]] = None
 
 class DocumentResponse(BaseModel):
     id: int
     name: str
     file_type: str
+    content_type: Optional[str] = None
     size: int
-    content_type: str
     created_at: datetime
     updated_at: datetime
     vector_database_id: int
     vector_database_name: Optional[str] = None
     is_embedded: bool
-    file_metadata: Optional[Dict[str, Any]] = None
     
     model_config = ConfigDict(from_attributes=True)
 
@@ -2261,15 +2375,32 @@ async def get_documents(
         # Add vector database name
         result = []
         for doc in documents:
-            doc_dict = DocumentResponse.model_validate(doc, from_attributes=True)
+            # Create a dictionary from the document for easier manipulation
+            doc_dict = {
+                "id": doc.id,
+                "name": doc.name,
+                "file_type": doc.file_type,
+                "content_type": doc.content_type,
+                "size": doc.size,
+                "created_at": doc.created_at,
+                "updated_at": doc.updated_at,
+                "vector_database_id": doc.vector_database_id or 0,  # Handle NULL values
+                "is_embedded": doc.is_embedded
+            }
             
             # Get vector database name if not already populated
-            if not hasattr(doc, 'vector_database_name') or doc.vector_database_name is None:
+            vector_db_name = None
+            if doc.vector_database_id is not None:
                 vector_db = db.query(VectorDatabase).filter(VectorDatabase.id == doc.vector_database_id).first()
                 vector_db_name = vector_db.name if vector_db else f"db_{doc.vector_database_id}"
-                doc_dict.vector_database_name = vector_db_name
+            else:
+                vector_db_name = "No Database"
+                
+            doc_dict["vector_database_name"] = vector_db_name
                 
-            result.append(doc_dict)
+            # Create Pydantic model from dictionary
+            doc_response = DocumentResponse(**doc_dict)
+            result.append(doc_response)
         
         return result
     except SQLAlchemyError as e:
@@ -2309,141 +2440,41 @@ async def get_document(
         logger.error(traceback.format_exc())
         raise HTTPException(status_code=500, detail=f"Error retrieving document: {str(e)}")
 
-@router.put("/documents/{document_id}", response_model=DocumentResponse)
-async def update_document(
-    document_id: int = Path(..., gt=0),
-    document_update: DocumentUpdate = Body(...),
-    db: Session = Depends(get_db)
-):
-    """
-    Update document details.
-    """
-    try:
-        document = db.query(Document).filter(Document.id == document_id).first()
-        if not document:
-            raise HTTPException(status_code=404, detail=f"Document with ID {document_id} not found")
-        
-        # Update fields if provided
-        if document_update.name is not None:
-            document.name = document_update.name
-            
-        if document_update.file_metadata is not None:
-            # Merge with existing metadata if it exists
-            if document.file_metadata:
-                document.file_metadata.update(document_update.file_metadata)
-            else:
-                document.file_metadata = document_update.file_metadata
-        
-        db.commit()
-        db.refresh(document)
-        
-        # Get vector database name
-        vector_db = db.query(VectorDatabase).filter(VectorDatabase.id == document.vector_database_id).first()
-        vector_db_name = vector_db.name if vector_db else f"db_{document.vector_database_id}"
-        
-        # Create response with vector database name
-        result = DocumentResponse.model_validate(document, from_attributes=True)
-        result.vector_database_name = vector_db_name
-        
-        return result
-    except HTTPException:
-        raise
-    except SQLAlchemyError as e:
-        db.rollback()
-        logger.error(f"Database error updating document: {e}")
-        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
-    except Exception as e:
-        db.rollback()
-        logger.error(f"Error updating document: {e}")
-        logger.error(traceback.format_exc())
-        raise HTTPException(status_code=500, detail=f"Error updating document: {str(e)}")
-
-@router.delete("/documents/{document_id}", response_model=dict)
-async def delete_document(
+@router.get("/documents/{document_id}/content", response_class=Response)
+async def get_document_content(
     document_id: int = Path(..., gt=0),
     db: Session = Depends(get_db)
 ):
     """
-    Delete document.
+    Get document content (file) by document ID.
+    Returns the binary content with the appropriate Content-Type header.
     """
     try:
+        # Get document to check if it exists and get metadata
         document = db.query(Document).filter(Document.id == document_id).first()
         if not document:
             raise HTTPException(status_code=404, detail=f"Document with ID {document_id} not found")
         
-        # Delete associated vector statuses
-        db.query(VectorStatus).filter(VectorStatus.document_id == document_id).delete()
-        
-        # Delete document
-        db.delete(document)
-        db.commit()
-        
-        return {"message": f"Document with ID {document_id} deleted successfully"}
-    except HTTPException:
-        raise
-    except SQLAlchemyError as e:
-        db.rollback()
-        logger.error(f"Database error deleting document: {e}")
-        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
-    except Exception as e:
-        db.rollback()
-        logger.error(f"Error deleting document: {e}")
-        logger.error(traceback.format_exc())
-        raise HTTPException(status_code=500, detail=f"Error deleting document: {str(e)}")
-
-@router.get("/vector-databases/{vector_db_id}/documents", response_model=List[DocumentResponse])
-async def get_documents_by_vector_db(
-    vector_db_id: int = Path(..., gt=0),
-    skip: int = 0,
-    limit: int = 100,
-    is_embedded: Optional[bool] = None,
-    file_type: Optional[str] = None,
-    db: Session = Depends(get_db)
-):
-    """
-    Get all documents for a specific vector database.
-    
-    - **skip**: Number of items to skip
-    - **limit**: Maximum number of items to return
-    - **is_embedded**: Filter by embedding status
-    - **file_type**: Filter by file type
-    """
-    try:
-        # Verify vector database exists
-        vector_db = db.query(VectorDatabase).filter(VectorDatabase.id == vector_db_id).first()
-        if not vector_db:
-            raise HTTPException(status_code=404, detail=f"Vector database with ID {vector_db_id} not found")
-        
-        # Build query
-        query = db.query(Document).filter(Document.vector_database_id == vector_db_id)
+        # Get document content from document_content table
+        document_content = db.query(DocumentContent).filter(DocumentContent.document_id == document_id).first()
+        if not document_content or not document_content.file_content:
+            raise HTTPException(status_code=404, detail=f"Content for document with ID {document_id} not found")
         
-        # Apply additional filters
-        if is_embedded is not None:
-            query = query.filter(Document.is_embedded == is_embedded)
-            
-        if file_type is not None:
-            query = query.filter(Document.file_type == file_type)
-        
-        # Execute query with pagination
-        documents = query.offset(skip).limit(limit).all()
+        # Determine content type
+        content_type = document.content_type if hasattr(document, 'content_type') and document.content_type else "application/octet-stream"
         
-        # Prepare results with vector database name
-        result = []
-        for doc in documents:
-            doc_response = DocumentResponse.model_validate(doc, from_attributes=True)
-            doc_response.vector_database_name = vector_db.name
-            result.append(doc_response)
-        
-        return result
+        # Return binary content with correct content type
+        return Response(
+            content=document_content.file_content,
+            media_type=content_type,
+            headers={"Content-Disposition": f"attachment; filename=\"{document.name}\""}
+        )
     except HTTPException:
         raise
-    except SQLAlchemyError as e:
-        logger.error(f"Database error retrieving documents: {e}")
-        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
     except Exception as e:
-        logger.error(f"Error retrieving documents: {e}")
+        logger.error(f"Error retrieving document content: {e}")
         logger.error(traceback.format_exc())
-        raise HTTPException(status_code=500, detail=f"Error retrieving documents: {str(e)}") 
+        raise HTTPException(status_code=500, detail=f"Error retrieving document content: {str(e)}")
 
 # --- Telegram Bot models and endpoints ---
 class TelegramBotBase(BaseModel):
@@ -2468,73 +2499,6 @@ class TelegramBotResponse(TelegramBotBase):
     
     model_config = ConfigDict(from_attributes=True)
 
-@router.get("/telegram-bots", response_model=List[TelegramBotResponse])
-async def get_telegram_bots(
-    skip: int = 0,
-    limit: int = 100,
-    status: Optional[str] = None,
-    db: Session = Depends(get_db)
-):
-    """
-    Get all Telegram bots.
-    
-    - **skip**: Number of items to skip
-    - **limit**: Maximum number of items to return
-    - **status**: Filter by status (e.g., 'active', 'inactive')
-    """
-    try:
-        query = db.query(TelegramBot)
-        
-        if status:
-            query = query.filter(TelegramBot.status == status)
-        
-        bots = query.offset(skip).limit(limit).all()
-        return [TelegramBotResponse.model_validate(bot, from_attributes=True) for bot in bots]
-    except SQLAlchemyError as e:
-        logger.error(f"Database error retrieving Telegram bots: {e}")
-        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
-    except Exception as e:
-        logger.error(f"Error retrieving Telegram bots: {e}")
-        logger.error(traceback.format_exc())
-        raise HTTPException(status_code=500, detail=f"Error retrieving Telegram bots: {str(e)}")
-
-@router.post("/telegram-bots", response_model=TelegramBotResponse)
-async def create_telegram_bot(
-    bot: TelegramBotCreate,
-    db: Session = Depends(get_db)
-):
-    """
-    Create a new Telegram bot.
-    """
-    try:
-        # Check if bot with this username already exists
-        existing_bot = db.query(TelegramBot).filter(TelegramBot.username == bot.username).first()
-        if existing_bot:
-            raise HTTPException(
-                status_code=400, 
-                detail=f"Telegram bot with username '{bot.username}' already exists"
-            )
-        
-        # Create new bot
-        db_bot = TelegramBot(**bot.model_dump())
-        
-        db.add(db_bot)
-        db.commit()
-        db.refresh(db_bot)
-        
-        return TelegramBotResponse.model_validate(db_bot, from_attributes=True)
-    except HTTPException:
-        raise
-    except SQLAlchemyError as e:
-        db.rollback()
-        logger.error(f"Database error creating Telegram bot: {e}")
-        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
-    except Exception as e:
-        db.rollback()
-        logger.error(f"Error creating Telegram bot: {e}")
-        logger.error(traceback.format_exc())
-        raise HTTPException(status_code=500, detail=f"Error creating Telegram bot: {str(e)}")
-
 @router.get("/telegram-bots/{bot_id}", response_model=TelegramBotResponse)
 async def get_telegram_bot(
     bot_id: int = Path(..., gt=0),
@@ -3543,4 +3507,301 @@ async def batch_delete_emergency_contacts(
         db.rollback()
         logger.error(f"Database error in batch_delete_emergency_contacts: {e}")
         logger.error(traceback.format_exc())
-        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
+
+@router.post("/documents", response_model=DocumentResponse)
+async def upload_document(
+    name: str = Form(...),
+    vector_database_id: int = Form(...),
+    file: UploadFile = File(...),
+    db: Session = Depends(get_db)
+):
+    """
+    Upload a new document and associate it with a vector database.
+    
+    - **name**: Document name
+    - **vector_database_id**: ID of the vector database to associate with
+    - **file**: The file to upload
+    """
+    try:
+        # Check if vector database exists
+        vector_db = db.query(VectorDatabase).filter(VectorDatabase.id == vector_database_id).first()
+        if not vector_db:
+            raise HTTPException(status_code=404, detail=f"Vector database with ID {vector_database_id} not found")
+        
+        # Read file content
+        file_content = await file.read()
+        file_size = len(file_content)
+        
+        # Determine file type from extension
+        filename = file.filename
+        file_extension = pathlib_Path(filename).suffix.lower()[1:] if filename else ""
+        
+        # Create document record
+        document = Document(
+            name=name,
+            vector_database_id=vector_database_id,
+            file_type=file_extension,
+            content_type=file.content_type,
+            size=file_size,
+            is_embedded=False
+        )
+        
+        db.add(document)
+        db.flush()  # Get ID without committing
+        
+        # Create document content record
+        document_content = DocumentContent(
+            document_id=document.id,
+            file_content=file_content
+        )
+        
+        db.add(document_content)
+        db.commit()
+        db.refresh(document)
+        
+        # Create vector status record for tracking embedding
+        vector_status = VectorStatus(
+            document_id=document.id,
+            vector_database_id=vector_database_id,
+            status="pending"
+        )
+        
+        db.add(vector_status)
+        db.commit()
+        
+        # Get vector database name for response
+        vector_db_name = vector_db.name if vector_db else f"db_{vector_database_id}"
+        
+        # Create response
+        result = DocumentResponse(
+            id=document.id,
+            name=document.name,
+            file_type=document.file_type,
+            content_type=document.content_type,
+            size=document.size,
+            created_at=document.created_at,
+            updated_at=document.updated_at,
+            vector_database_id=document.vector_database_id,
+            vector_database_name=vector_db_name,
+            is_embedded=document.is_embedded
+        )
+        
+        return result
+    except HTTPException:
+        raise
+    except SQLAlchemyError as e:
+        db.rollback()
+        logger.error(f"Database error uploading document: {e}")
+        logger.error(traceback.format_exc())
+        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
+    except Exception as e:
+        db.rollback()
+        logger.error(f"Error uploading document: {e}")
+        logger.error(traceback.format_exc())
+        raise HTTPException(status_code=500, detail=f"Error uploading document: {str(e)}")
+
+@router.put("/documents/{document_id}", response_model=DocumentResponse)
+async def update_document(
+    document_id: int,
+    name: Optional[str] = Form(None),
+    file: Optional[UploadFile] = File(None),
+    background_tasks: BackgroundTasks = None,
+    db: Session = Depends(get_db)
+):
+    """
+    Update an existing document. Can update name, file content, or both.
+    
+    - **document_id**: ID of the document to update
+    - **name**: New document name (optional)
+    - **file**: New file content (optional)
+    """
+    try:
+        # Validate document_id
+        if document_id <= 0:
+            raise HTTPException(status_code=400, detail="document_id must be greater than 0")
+            
+        # Check if document exists
+        document = db.query(Document).filter(Document.id == document_id).first()
+        if not document:
+            raise HTTPException(status_code=404, detail=f"Document with ID {document_id} not found")
+        
+        # Get vector database information for later use
+        vector_db = None
+        if document.vector_database_id:
+            vector_db = db.query(VectorDatabase).filter(VectorDatabase.id == document.vector_database_id).first()
+        
+        # Update name if provided
+        if name:
+            document.name = name
+        
+        # Update file if provided
+        if file:
+            # Read new file content
+            file_content = await file.read()
+            file_size = len(file_content)
+            
+            # Determine file type from extension
+            filename = file.filename
+            file_extension = pathlib_Path(filename).suffix.lower()[1:] if filename else ""
+            
+            # Update document record
+            document.file_type = file_extension
+            document.content_type = file.content_type
+            document.size = file_size
+            document.is_embedded = False  # Reset embedding status
+            document.updated_at = datetime.now()
+            
+            # Update document content
+            document_content = db.query(DocumentContent).filter(DocumentContent.document_id == document_id).first()
+            if document_content:
+                document_content.file_content = file_content
+            else:
+                # Create new document content if it doesn't exist
+                document_content = DocumentContent(
+                    document_id=document_id,
+                    file_content=file_content
+                )
+                db.add(document_content)
+            
+            # Get vector status for Pinecone cleanup
+            vector_status = db.query(VectorStatus).filter(VectorStatus.document_id == document_id).first()
+            
+            # Store old vector_id for cleanup
+            old_vector_id = None
+            if vector_status and vector_status.vector_id:
+                old_vector_id = vector_status.vector_id
+            
+            # Update vector status to pending
+            if vector_status:
+                vector_status.status = "pending"
+                vector_status.vector_id = None
+                vector_status.embedded_at = None
+                vector_status.error_message = None
+            else:
+                # Create new vector status if it doesn't exist
+                vector_status = VectorStatus(
+                    document_id=document_id,
+                    vector_database_id=document.vector_database_id,
+                    status="pending"
+                )
+                db.add(vector_status)
+            
+            # Schedule deletion of old vectors in Pinecone if we have all needed info
+            if old_vector_id and vector_db and document.vector_database_id and background_tasks:
+                try:
+                    # Initialize PDFProcessor for vector deletion
+                    from app.pdf.processor import PDFProcessor
+                    
+                    processor = PDFProcessor(
+                        index_name=vector_db.pinecone_index,
+                        namespace=f"vdb-{document.vector_database_id}",
+                        vector_db_id=document.vector_database_id
+                    )
+                    
+                    # Add deletion task to background tasks
+                    background_tasks.add_task(
+                        processor.delete_document_vectors,
+                        old_vector_id
+                    )
+                    
+                    logger.info(f"Scheduled deletion of old vectors for document {document_id}")
+                except Exception as e:
+                    logger.error(f"Error scheduling vector deletion: {str(e)}")
+                    # Continue with the update even if vector deletion scheduling fails
+            
+            # Schedule document for re-embedding if possible
+            if background_tasks and document.vector_database_id:
+                try:
+                    # Import here to avoid circular imports
+                    from app.pdf.tasks import process_document_for_embedding
+                    
+                    # Schedule embedding
+                    background_tasks.add_task(
+                        process_document_for_embedding,
+                        document_id=document_id,
+                        vector_db_id=document.vector_database_id
+                    )
+                    
+                    logger.info(f"Scheduled re-embedding for document {document_id}")
+                except Exception as e:
+                    logger.error(f"Error scheduling document embedding: {str(e)}")
+                    # Continue with the update even if embedding scheduling fails
+        
+        db.commit()
+        db.refresh(document)
+        
+        # Get vector database name for response
+        vector_db_name = "No Database"
+        if vector_db:
+            vector_db_name = vector_db.name
+        elif document.vector_database_id:
+            vector_db_name = f"db_{document.vector_database_id}"
+        
+        # Create response
+        result = DocumentResponse(
+            id=document.id,
+            name=document.name,
+            file_type=document.file_type,
+            content_type=document.content_type,
+            size=document.size,
+            created_at=document.created_at,
+            updated_at=document.updated_at,
+            vector_database_id=document.vector_database_id or 0,
+            vector_database_name=vector_db_name,
+            is_embedded=document.is_embedded
+        )
+        
+        return result
+    except HTTPException:
+        raise
+    except SQLAlchemyError as e:
+        db.rollback()
+        logger.error(f"Database error updating document: {e}")
+        logger.error(traceback.format_exc())
+        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
+    except Exception as e:
+        db.rollback()
+        logger.error(f"Error updating document: {e}")
+        logger.error(traceback.format_exc())
+        raise HTTPException(status_code=500, detail=f"Error updating document: {str(e)}")
+
+@router.delete("/documents/{document_id}", response_model=dict)
+async def delete_document(
+    document_id: int = Path(..., gt=0),
+    db: Session = Depends(get_db)
+):
+    """
+    Delete a document and its associated content.
+    
+    - **document_id**: ID of the document to delete
+    """
+    try:
+        # Check if document exists
+        document = db.query(Document).filter(Document.id == document_id).first()
+        if not document:
+            raise HTTPException(status_code=404, detail=f"Document with ID {document_id} not found")
+        
+        # Delete vector status
+        db.query(VectorStatus).filter(VectorStatus.document_id == document_id).delete()
+        
+        # Delete document content
+        db.query(DocumentContent).filter(DocumentContent.document_id == document_id).delete()
+        
+        # Delete document
+        db.delete(document)
+        db.commit()
+        
+        return {"status": "success", "message": f"Document with ID {document_id} deleted successfully"}
+    except HTTPException:
+        raise
+    except SQLAlchemyError as e:
+        db.rollback()
+        logger.error(f"Database error deleting document: {e}")
+        logger.error(traceback.format_exc())
+        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
+    except Exception as e:
+        db.rollback()
+        logger.error(f"Error deleting document: {e}")
+        logger.error(traceback.format_exc())
+        raise HTTPException(status_code=500, detail=f"Error deleting document: {str(e)}")

app/api/rag_routes.py

@@ -48,17 +48,17 @@ router = APIRouter(
 
 fix_request = PromptTemplate(
     template = """Goal:
-Your task is fixing user'srequest to get all information of history chat.
-You will received a conversation history and current request of user.
-Generate a new request that make sense if current request related to history conversation.
+Your task is to extract important keywords from the user's current request, optionally using chat history if relevant.
+You will receive a conversation history and the user's current message.
+Generate a **list of concise keywords** that best represent the user's intent.
 
 Return Format:
-Only return the fully users' request with all the important keywords.
-If the current message is NOT related to the conversation history or there is no chat history: Return user's current request.
-If the current message IS related to the conversation history: Return new request based on information from the conversation history and the current request.
+Only return keywords (comma-separated, no extra explanation).
+If the current message is NOT related to the chat history or if there is no chat history: Return keywords from the current message only.
+If the current message IS related to the chat history: Return a refined set of keywords based on both history and current message.
 
 Warning:
-Only use history chat if current request is truly relevant to the previous conversation.
+Only use chat history if the current message is clearly related to the prior context.
 
 Conversation History:
 {chat_history}
@@ -66,15 +66,24 @@ Conversation History:
 User current message:
 {question}
 """,
-    input_variables = ["chat_history", "question"],
+    input_variables=["chat_history", "question"],
 )
 
 # Create a prompt template with conversation history
 prompt = PromptTemplate(
     template = """Goal:
-You are a professional tour guide assistant that assists users in finding information about places in Da Nang, Vietnam.
+You are Pixity - a professional tour guide assistant that assists users in finding information about places in Da Nang, Vietnam.
 You can provide details on restaurants, cafes, hotels, attractions, and other local venues. 
 You have to use core knowledge and conversation history to chat with users, who are Da Nang's tourists. 
+Pixity’s Core Personality: Friendly & Warm: Chats like a trustworthy friend who listens and is always ready to help.
+Naturally Cute: Shows cuteness through word choice, soft emojis, and gentle care for the user.
+Playful – a little bit cheeky in a lovable way: Occasionally cracks jokes, uses light memes or throws in a surprise response that makes users smile. Think Duolingo-style humor, but less threatening.
+Smart & Proactive: Friendly, but also delivers quick, accurate info. Knows how to guide users to the right place – at the right time – with the right solution.
+Tone & Voice: Friendly – Youthful – Snappy. Uses simple words, similar to daily chat language (e.g., “Let’s find it together!” / “Need a tip?” / “Here’s something cool”). Avoids sounding robotic or overly scripted. Can joke lightly in smart ways, making Pixity feel like a travel buddy who knows how to lift the mood
+SAMPLE DIALOGUES
+When a user opens the chatbot for the first time:
+User: Hello?
+Pixity: Hi hi 👋 I’ve been waiting for you! Ready to explore Da Nang together? I’ve got tips, tricks, and a tiny bit of magic 🎒✨
 
 Return Format:
 Respond in friendly, natural, concise and use only English like a real tour guide.
@@ -251,7 +260,7 @@ async def chat(request: ChatRequest, background_tasks: BackgroundTasks):
         # Generate the prompt using template
         prompt_text = prompt.format(
             context=context,
-            question=final_request.text,
+            question=request.question,
             chat_history=chat_history
         )
         logger.info(f"Full prompt with history and context: {prompt_text}")

app/api/websocket_routes.py

@@ -92,7 +92,7 @@ def get_full_websocket_url(server_side=False):
     3. When there are new sessions requiring attention, you will receive notifications through this connection
     
     Notifications are sent when:
-    - Session response starts with "I don't know"
+    - Session response starts with "I'm sorry"
     - The system cannot answer the user's question
     
     Make sure to send a "keepalive" message every 5 minutes to maintain the connection.
@@ -114,14 +114,14 @@ async def websocket_documentation():
         "full_url": ws_url,
         "description": "Endpoint to receive notifications about new sessions requiring attention",
         "notification_format": {
-            "type": "new_session",
+            "type": "sorry_response",
             "timestamp": "YYYY-MM-DD HH:MM:SS",
             "data": {
                 "session_id": "session id",
                 "factor": "user",
                 "action": "action type",
                 "message": "User question",
-                "response": "I don't know...",
+                "response": "I'm sorry...",
                 "user_id": "user id",
                 "first_name": "user's first name",
                 "last_name": "user's last name",
@@ -168,7 +168,7 @@ async def websocket_documentation():
                 data = json.loads(message)
                 print(f"Received notification: {data}")
                 # Process notification, e.g.: send to Telegram Admin
-                if data.get("type") == "new_session":
+                if data.get("type") == "sorry_response":
                     session_data = data.get("data", {})
                     user_question = session_data.get("message", "")
                     user_name = session_data.get("first_name", "Unknown User")
@@ -230,18 +230,60 @@ async def websocket_endpoint(websocket: WebSocket):
     """
     await manager.connect(websocket)
     try:
+        # Keep track of last activity time to prevent connection timeouts
+        last_activity = datetime.now()
+        
+        # Set up a background ping task
+        async def send_periodic_ping():
+            try:
+                while True:
+                    # Send ping every 20 seconds if no other activity
+                    await asyncio.sleep(20)
+                    current_time = datetime.now()
+                    time_since_activity = (current_time - last_activity).total_seconds()
+                    
+                    # Only send ping if there's been no activity for 15+ seconds
+                    if time_since_activity > 15:
+                        logger.debug("Sending ping to client to keep connection alive")
+                        await websocket.send_json({"type": "ping", "timestamp": current_time.isoformat()})
+            except asyncio.CancelledError:
+                # Task was cancelled, just exit quietly
+                pass
+            except Exception as e:
+                logger.error(f"Error in ping task: {e}")
+        
+        # Start ping task
+        ping_task = asyncio.create_task(send_periodic_ping())
+        
+        # Main message loop
         while True:
+            # Update last activity time
+            last_activity = datetime.now()
+            
             # Maintain WebSocket connection
             data = await websocket.receive_text()
+            
             # Echo back to keep connection active
-            await websocket.send_json({"status": "connected", "echo": data, "timestamp": datetime.now().isoformat()})
+            await websocket.send_json({
+                "status": "connected", 
+                "echo": data, 
+                "timestamp": last_activity.isoformat()
+            })
             logger.info(f"Received message from WebSocket: {data}")
     except WebSocketDisconnect:
         logger.info("WebSocket client disconnected")
-        manager.disconnect(websocket)
     except Exception as e:
         logger.error(f"WebSocket error: {e}")
+    finally:
+        # Always clean up properly
         manager.disconnect(websocket)
+        # Cancel ping task if it's still running
+        try:
+            ping_task.cancel()
+            await ping_task
+        except (UnboundLocalError, asyncio.CancelledError):
+            # ping_task wasn't created or already cancelled
+            pass
 
 # Function to send notifications over WebSocket
 async def send_notification(data: dict):
@@ -249,7 +291,7 @@ async def send_notification(data: dict):
     Send notification to all active WebSocket connections.
     
     This function is used to notify admin bots about new issues or questions that need attention.
-    It's triggered when the system cannot answer a user's question (response starts with "I don't know").
+    It's triggered when the system cannot answer a user's question (response starts with "I'm sorry").
     
     Args:
         data: The data to send as notification
@@ -260,33 +302,30 @@ async def send_notification(data: dict):
         logger.info(f"Notification data: session_id={data.get('session_id')}, user_id={data.get('user_id')}")
         logger.info(f"Response: {data.get('response', '')[:50]}...")
         
-        # Check if the response starts with "I don't know"
+        # Check if the response starts with "I'm sorry"
         response = data.get('response', '')
         if not response or not isinstance(response, str):
             logger.warning(f"Invalid response format in notification data: {response}")
             return
             
-        if not response.strip().lower().startswith("i don't know"):
-            logger.info(f"Response doesn't start with 'I don't know', notification not needed: {response[:50]}...")
+        if not response.strip().lower().startswith("i'm sorry"):
+            logger.info(f"Response doesn't start with 'I'm sorry', notification not needed: {response[:50]}...")
             return
             
-        logger.info(f"Response starts with 'I don't know', sending notification")
+        logger.info(f"Response starts with 'I'm sorry', sending notification")
         
-        # Format the notification data for admin
+        # Format the notification data for admin - format theo chuẩn Admin_bot
         notification_data = {
-            "type": "new_session",
+            "type": "sorry_response",  # Đổi type thành sorry_response để phù hợp với Admin_bot
             "timestamp": get_local_time(),
-            "data": {
-                "session_id": data.get('session_id', 'unknown'),
-                "user_id": data.get('user_id', 'unknown'),
-                "message": data.get('message', ''),
-                "response": response,
+            "user_id": data.get('user_id', 'unknown'),
+            "message": data.get('message', ''),
+            "response": response,
+            "session_id": data.get('session_id', 'unknown'),
+            "user_info": {
                 "first_name": data.get('first_name', 'User'),
                 "last_name": data.get('last_name', ''),
-                "username": data.get('username', ''),
-                "created_at": data.get('created_at', get_local_time()),
-                "action": data.get('action', 'unknown'),
-                "factor": "user"  # Always show as user for better readability
+                "username": data.get('username', '')
             }
         }
         

app/database/models.py

@@ -78,7 +78,7 @@ class VectorDatabase(Base):
     name = Column(String, nullable=False, unique=True)
     description = Column(String, nullable=True)
     pinecone_index = Column(String, nullable=False)
-    api_key = Column(String, nullable=False)
+    api_key_id = Column(Integer, ForeignKey("api_key.id"), nullable=True)
     status = Column(String, default="active")
     created_at = Column(DateTime, server_default=func.now())
     updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
@@ -87,18 +87,17 @@ class VectorDatabase(Base):
     documents = relationship("Document", back_populates="vector_database")
     vector_statuses = relationship("VectorStatus", back_populates="vector_database")
     engine_associations = relationship("EngineVectorDb", back_populates="vector_database")
+    api_key_ref = relationship("ApiKey", foreign_keys=[api_key_id])
 
 class Document(Base):
     __tablename__ = "document"
     
     id = Column(Integer, primary_key=True, index=True)
     name = Column(String, nullable=False)
-    file_content = Column(LargeBinary, nullable=True)
     file_type = Column(String, nullable=True)
-    size = Column(Integer, nullable=True)
     content_type = Column(String, nullable=True)
+    size = Column(Integer, nullable=True)
     is_embedded = Column(Boolean, default=False)
-    file_metadata = Column(JSON, nullable=True)
     vector_database_id = Column(Integer, ForeignKey("vector_database.id"), nullable=False)
     created_at = Column(DateTime, server_default=func.now())
     updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
@@ -106,6 +105,18 @@ class Document(Base):
     # Relationships
     vector_database = relationship("VectorDatabase", back_populates="documents")
     vector_statuses = relationship("VectorStatus", back_populates="document")
+    file_content_ref = relationship("DocumentContent", back_populates="document", uselist=False, cascade="all, delete-orphan")
+
+class DocumentContent(Base):
+    __tablename__ = "document_content"
+    
+    id = Column(Integer, primary_key=True, index=True)
+    document_id = Column(Integer, ForeignKey("document.id"), nullable=False, unique=True)
+    file_content = Column(LargeBinary, nullable=True)
+    created_at = Column(DateTime, server_default=func.now())
+    
+    # Relationships
+    document = relationship("Document", back_populates="file_content_ref")
 
 class VectorStatus(Base):
     __tablename__ = "vector_status"
@@ -184,9 +195,10 @@ class ApiKey(Base):
     __tablename__ = "api_key"
     
     id = Column(Integer, primary_key=True, index=True)
-    key = Column(String, nullable=False, unique=True)
-    name = Column(String, nullable=False)
-    description = Column(String, nullable=True)
-    is_active = Column(Boolean, default=True)
+    key_type = Column(String, nullable=False)
+    key_value = Column(Text, nullable=False)
+    description = Column(Text, nullable=True)
     created_at = Column(DateTime, server_default=func.now())
-    last_used = Column(DateTime, nullable=True) 
+    last_used = Column(DateTime, nullable=True)
+    expires_at = Column(DateTime, nullable=True)
+    is_active = Column(Boolean, default=True) 

app/database/mongodb.py

@@ -142,13 +142,40 @@ def get_chat_history(user_id, n = 5) -> str:
     Bot: ...
     User: ...
     Bot: ...
+    
+    Chỉ lấy history sau lệnh /start hoặc /clear mới nhất
     """
     try:
-        # Truy vấn các document có user_id, sắp xếp theo created_at tăng dần
-        # Get the 4 most recent documents first, then sort them in ascending order
-        docs = list(session_collection.find({"user_id": str(user_id)}).sort("created_at", -1).limit(n))
-        # Reverse the list to get chronological order (oldest to newest)
-        docs.reverse()
+        # Tìm session /start hoặc /clear mới nhất
+        reset_session = session_collection.find_one(
+            {
+                "user_id": str(user_id), 
+                "$or": [
+                    {"action": "start"},
+                    {"action": "clear"}
+                ]
+            },
+            sort=[("created_at_datetime", -1)]
+        )
+        
+        # Nếu không tìm thấy session reset nào, lấy n session gần nhất
+        if reset_session:
+            reset_time = reset_session["created_at_datetime"]
+            # Lấy các session sau reset_time
+            docs = list(
+                session_collection.find({
+                    "user_id": str(user_id),
+                    "created_at_datetime": {"$gt": reset_time}
+                }).sort("created_at_datetime", 1)
+            )
+            logger.info(f"Lấy {len(docs)} session sau lệnh {reset_session['action']} lúc {reset_time}")
+        else:
+            # Không tìm thấy reset session, lấy n session gần nhất
+            docs = list(session_collection.find({"user_id": str(user_id)}).sort("created_at", -1).limit(n))
+            # Đảo ngược để có thứ tự từ cũ đến mới
+            docs.reverse()
+            logger.info(f"Không tìm thấy session reset, lấy {len(docs)} session gần nhất")
+            
         if not docs:
             logger.info(f"Không tìm thấy dữ liệu cho user_id: {user_id}")
             return ""
@@ -161,6 +188,10 @@ def get_chat_history(user_id, n = 5) -> str:
             message = doc.get("message", "")
             response = doc.get("response", "")
             
+            # Bỏ qua lệnh start và clear
+            if action in ["start", "clear"]:
+                continue
+                
             if factor == "user" and action == "asking_freely":
                 conversation_lines.append(f"User: {message}")
                 conversation_lines.append(f"Bot: {response}")
@@ -174,13 +205,14 @@ def get_chat_history(user_id, n = 5) -> str:
 def get_request_history(user_id, n=3):
     """Get the most recent user requests to use as context for retrieval"""
     try:
-        # Lấy lịch sử trực tiếp từ MongoDB (thông qua get_user_history đã sửa đổi)
-        history = get_user_history(user_id, n)
+        # Truy vấn trực tiếp từ MongoDB
+        history = get_chat_history(user_id, n)
         
         # Just extract the questions for context
         requests = []
-        for item in history:
-            requests.append(item['question'])
+        for line in history.split('\n'):
+            if line.startswith("User: "):
+                requests.append(line[6:])  # Lấy nội dung sau "User: "
             
         # Join all recent requests into a single string for context
         return " ".join(requests)

app/database/postgresql.py

@@ -30,11 +30,11 @@ if not DATABASE_URL:
 try:
     engine = create_engine(
         DATABASE_URL,
-        pool_pre_ping=True,         # Enable connection health checks
-        pool_recycle=300,           # Recycle connections every 5 minutes
-        pool_size=20,               # Increase pool size for more concurrent connections
-        max_overflow=30,            # Allow more overflow connections
-        pool_timeout=30,            # Timeout for getting connection from pool
+        pool_size=10,  # Limit max connections
+        max_overflow=5,  # Allow temporary overflow of connections
+        pool_timeout=30,  # Timeout waiting for connection from pool
+        pool_recycle=300,  # Recycle connections every 5 minutes
+        pool_pre_ping=True,  # Verify connection is still valid before using it
         connect_args={
             "connect_timeout": 5,   # Connection timeout in seconds
             "keepalives": 1,        # Enable TCP keepalives
@@ -89,18 +89,17 @@ def check_db_connection():
 
 # Dependency to get DB session with improved error handling
 def get_db():
-    """Get database session dependency for FastAPI endpoints"""
+    """Get PostgreSQL database session"""
     db = SessionLocal()
     try:
-        # Test connection is valid before returning
+        # Test connection
         db.execute(text("SELECT 1")).fetchone()
         yield db
-    except SQLAlchemyError as e:
-        logger.error(f"Database session error: {e}")
-        db.rollback()
+    except Exception as e:
+        logger.error(f"DB connection error: {e}")
         raise
     finally:
-        db.close()
+        db.close()  # Ensure connection is closed and returned to pool
 
 # Create tables in database if they don't exist
 def create_tables():

app/models/pdf_models.py

@@ -7,14 +7,19 @@ class PDFUploadRequest(BaseModel):
     index_name: Optional[str] = Field("testbot768", description="Tên index trong Pinecone")
     title: Optional[str] = Field(None, description="Tiêu đề của tài liệu")
     description: Optional[str] = Field(None, description="Mô tả về tài liệu")
+    vector_database_id: Optional[int] = Field(None, description="ID của vector database trong PostgreSQL để sử dụng")
 
 class PDFResponse(BaseModel):
-    """Response model cho xử lý PDF"""
-    success: bool = Field(..., description="Trạng thái xử lý thành công hay không")
-    document_id: Optional[str] = Field(None, description="ID của tài liệu")
+    """Response model cho các endpoints liên quan đến PDF."""
+    success: bool = Field(False, description="Kết quả xử lý: true/false")
+    document_id: Optional[str] = Field(None, description="ID của tài liệu đã xử lý")
+    document_database_id: Optional[int] = Field(None, description="ID của tài liệu trong PostgreSQL (nếu có)")
     chunks_processed: Optional[int] = Field(None, description="Số lượng chunks đã xử lý")
-    total_text_length: Optional[int] = Field(None, description="Tổng độ dài văn bản")
-    error: Optional[str] = Field(None, description="Thông báo lỗi nếu có")
+    total_text_length: Optional[int] = Field(None, description="Tổng kích thước text đã xử lý")
+    error: Optional[str] = Field(None, description="Thông báo lỗi (nếu có)")
+    warning: Optional[str] = Field(None, description="Cảnh báo (nếu có)")
+    mock_mode: Optional[bool] = Field(None, description="Đã chạy ở chế độ mock hay không")
+    message: Optional[str] = Field(None, description="Thông báo thành công")
     
     class Config:
         schema_extra = {
@@ -22,7 +27,9 @@ class PDFResponse(BaseModel):
                 "success": True,
                 "document_id": "550e8400-e29b-41d4-a716-446655440000",
                 "chunks_processed": 25,
-                "total_text_length": 50000
+                "total_text_length": 50000,
+                "mock_mode": False,
+                "message": "Successfully processed document"
             }
         }
 
@@ -31,14 +38,18 @@ class DeleteDocumentRequest(BaseModel):
     document_id: str = Field(..., description="ID của tài liệu cần xóa")
     namespace: Optional[str] = Field("Default", description="Namespace trong Pinecone")
     index_name: Optional[str] = Field("testbot768", description="Tên index trong Pinecone")
+    vector_database_id: Optional[int] = Field(None, description="ID của vector database trong PostgreSQL")
 
 class DocumentsListResponse(BaseModel):
-    """Response model cho lấy danh sách tài liệu"""
-    success: bool = Field(..., description="Trạng thái xử lý thành công hay không")
-    total_vectors: Optional[int] = Field(None, description="Tổng số vectors trong index")
-    namespace: Optional[str] = Field(None, description="Namespace đang sử dụng")
-    index_name: Optional[str] = Field(None, description="Tên index đang sử dụng")
-    error: Optional[str] = Field(None, description="Thông báo lỗi nếu có")
+    """Response model cho danh sách documents"""
+    success: bool = Field(False, description="Kết quả xử lý: true/false")
+    total_vectors: Optional[int] = Field(None, description="Tổng số vectors trong namespace")
+    namespace: Optional[str] = Field(None, description="Namespace đã truy vấn")
+    index_name: Optional[str] = Field(None, description="Tên index đã truy vấn")
+    documents: Optional[List[Dict[str, Any]]] = Field(None, description="Danh sách documents")
+    postgresql_documents: Optional[List[Dict[str, Any]]] = Field(None, description="Danh sách documents từ PostgreSQL")
+    postgresql_document_count: Optional[int] = Field(None, description="Số lượng documents từ PostgreSQL")
+    error: Optional[str] = Field(None, description="Thông báo lỗi (nếu có)")
     
     class Config:
         schema_extra = {

app/utils/cache.py

@@ -17,8 +17,6 @@ load_dotenv()
 DEFAULT_CACHE_TTL = int(os.getenv("CACHE_TTL_SECONDS", "300"))  # Mặc định 5 phút
 DEFAULT_CACHE_CLEANUP_INTERVAL = int(os.getenv("CACHE_CLEANUP_INTERVAL", "60"))  # Mặc định 1 phút
 DEFAULT_CACHE_MAX_SIZE = int(os.getenv("CACHE_MAX_SIZE", "1000"))  # Mặc định 1000 phần tử
-DEFAULT_HISTORY_QUEUE_SIZE = int(os.getenv("HISTORY_QUEUE_SIZE", "10"))  # Mặc định queue size là 10
-DEFAULT_HISTORY_CACHE_TTL = int(os.getenv("HISTORY_CACHE_TTL", "3600"))  # Mặc định 1 giờ
 
 # Generic type để có thể sử dụng cho nhiều loại giá trị khác nhau
 T = TypeVar('T')
@@ -42,36 +40,6 @@ class CacheItem(Generic[T]):
         """Gia hạn thời gian sống của item"""
         self.expire_at = time.time() + ttl
 
-
-# Lớp HistoryQueue để lưu trữ lịch sử người dùng
-class HistoryQueue:
-    def __init__(self, max_size: int = DEFAULT_HISTORY_QUEUE_SIZE, ttl: int = DEFAULT_HISTORY_CACHE_TTL):
-        self.items: List[Dict[str, Any]] = []
-        self.max_size = max_size
-        self.ttl = ttl
-        self.expire_at = time.time() + ttl
-    
-    def add(self, item: Dict[str, Any]) -> None:
-        """Thêm một item vào queue, nếu đã đầy thì loại bỏ item cũ nhất"""
-        if len(self.items) >= self.max_size:
-            self.items.pop(0)
-        self.items.append(item)
-        # Mỗi khi thêm item mới, cập nhật thời gian hết hạn
-        self.refresh_expiry()
-    
-    def get_all(self) -> List[Dict[str, Any]]:
-        """Lấy tất cả items trong queue"""
-        return self.items
-    
-    def is_expired(self) -> bool:
-        """Kiểm tra xem queue có hết hạn chưa"""
-        return time.time() > self.expire_at
-    
-    def refresh_expiry(self) -> None:
-        """Làm mới thời gian hết hạn"""
-        self.expire_at = time.time() + self.ttl
-
-
 # Lớp cache chính
 class InMemoryCache:
     def __init__(
@@ -84,7 +52,6 @@ class InMemoryCache:
         self.ttl = ttl
         self.cleanup_interval = cleanup_interval
         self.max_size = max_size
-        self.user_history_queues: Dict[str, HistoryQueue] = {}
         self.lock = threading.RLock()  # Sử dụng RLock để tránh deadlock
         
         # Khởi động thread dọn dẹp cache định kỳ (active expiration)
@@ -170,13 +137,8 @@ class InMemoryCache:
             for key in expired_keys:
                 del self.cache[key]
             
-            # Xóa các user history queue đã hết hạn
-            expired_user_ids = [uid for uid, queue in self.user_history_queues.items() if queue.is_expired()]
-            for user_id in expired_user_ids:
-                del self.user_history_queues[user_id]
-                
-            if expired_keys or expired_user_ids:
-                logger.debug(f"Cleaned up {len(expired_keys)} expired cache items and {len(expired_user_ids)} expired history queues")
+            if expired_keys:
+                logger.debug(f"Cleaned up {len(expired_keys)} expired cache items")
     
     def _evict_lru_items(self, count: int = 1) -> None:
         """Xóa bỏ các item ít được truy cập nhất khi cache đầy"""
@@ -198,8 +160,7 @@ class InMemoryCache:
                 "active_items": total_items - expired_items,
                 "memory_usage_bytes": memory_usage,
                 "memory_usage_mb": memory_usage / (1024 * 1024),
-                "max_size": self.max_size,
-                "history_queues": len(self.user_history_queues)
+                "max_size": self.max_size
             }
     
     def _estimate_memory_usage(self) -> int:
@@ -219,46 +180,7 @@ class InMemoryCache:
             except:
                 cache_size += 100
         
-        # Ước tính kích thước của user history queues
-        for queue in self.user_history_queues.values():
-            try:
-                cache_size += len(json.dumps(queue.items)) + 100  # 100 bytes cho metadata
-            except:
-                cache_size += 100
-        
         return cache_size
-    
-    # Các phương thức chuyên biệt cho việc quản lý lịch sử người dùng
-    def add_user_history(self, user_id: str, item: Dict[str, Any], queue_size: Optional[int] = None, ttl: Optional[int] = None) -> None:
-        """Thêm một item vào history queue của người dùng"""
-        with self.lock:
-            # Tạo queue nếu chưa tồn tại
-            if user_id not in self.user_history_queues:
-                queue_size_value = queue_size if queue_size is not None else DEFAULT_HISTORY_QUEUE_SIZE
-                ttl_value = ttl if ttl is not None else DEFAULT_HISTORY_CACHE_TTL
-                self.user_history_queues[user_id] = HistoryQueue(max_size=queue_size_value, ttl=ttl_value)
-            
-            # Thêm item vào queue
-            self.user_history_queues[user_id].add(item)
-            logger.debug(f"Added history item for user {user_id}")
-    
-    def get_user_history(self, user_id: str, default: Any = None) -> List[Dict[str, Any]]:
-        """Lấy lịch sử của người dùng từ cache"""
-        with self.lock:
-            queue = self.user_history_queues.get(user_id)
-            
-            # Nếu không tìm thấy queue hoặc queue đã hết hạn
-            if queue is None or queue.is_expired():
-                if queue is not None and queue.is_expired():
-                    del self.user_history_queues[user_id]
-                    logger.debug(f"User history queue expired: {user_id}")
-                return default if default is not None else []
-            
-            # Làm mới thời gian hết hạn
-            queue.refresh_expiry()
-            logger.debug(f"Retrieved history for user {user_id}: {len(queue.items)} items")
-            return queue.get_all()
-
 
 # Singleton instance
 _cache_instance = None

app/utils/pdf_processor.py

@@ -1,211 +1,449 @@
 import os
-import time
+import logging
 import uuid
-from langchain.text_splitter import RecursiveCharacterTextSplitter
+import pinecone
+from app.utils.pinecone_fix import PineconeConnectionManager, check_connection
+import time
+import os
+from typing import List, Dict, Any, Optional
+
+# Langchain imports for document processing
 from langchain_community.document_loaders import PyPDFLoader
+from langchain.text_splitter import RecursiveCharacterTextSplitter
 from langchain_google_genai import GoogleGenerativeAIEmbeddings
-import logging
+import google.generativeai as genai
 
-from app.database.pinecone import get_pinecone_index, init_pinecone
-
-# Cấu hình logging
+# Configure logger
 logger = logging.getLogger(__name__)
 
-# Khởi tạo embeddings model
-embeddings_model = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
-
 class PDFProcessor:
-    """Lớp xử lý file PDF và tạo embeddings"""
+    """Process PDF files and create embeddings in Pinecone"""
     
-    def __init__(self, index_name="testbot768", namespace="Default"):
-        """Khởi tạo với tên index và namespace Pinecone mặc định"""
+    def __init__(self, index_name="testbot768", namespace="Default", api_key=None, vector_db_id=None, mock_mode=False, correlation_id=None):
         self.index_name = index_name
         self.namespace = namespace
+        self.api_key = api_key
+        self.vector_db_id = vector_db_id
         self.pinecone_index = None
+        self.mock_mode = mock_mode
+        self.correlation_id = correlation_id or str(uuid.uuid4())[:8]
+        self.google_api_key = os.environ.get("GOOGLE_API_KEY")
         
-    def _init_pinecone_connection(self):
-        """Khởi tạo kết nối đến Pinecone"""
-        try:
-            # Sử dụng singleton pattern từ module database.pinecone
-            self.pinecone_index = get_pinecone_index()
-            if not self.pinecone_index:
-                logger.error("Không thể kết nối đến Pinecone")
-                return False
-            return True
-        except Exception as e:
-            logger.error(f"Lỗi khi kết nối Pinecone: {str(e)}")
-            return False
+        # Initialize Pinecone connection if not in mock mode
+        if not self.mock_mode and self.api_key:
+            try:
+                # Use connection manager from pinecone_fix
+                logger.info(f"[{self.correlation_id}] Initializing Pinecone connection to {self.index_name}")
+                self.pinecone_index = PineconeConnectionManager.get_index(self.api_key, self.index_name)
+                logger.info(f"[{self.correlation_id}] Successfully connected to Pinecone index {self.index_name}")
+            except Exception as e:
+                logger.error(f"[{self.correlation_id}] Failed to initialize Pinecone: {str(e)}")
+                # Fall back to mock mode if connection fails
+                self.mock_mode = True
+                logger.warning(f"[{self.correlation_id}] Falling back to mock mode due to connection error")
             
     async def process_pdf(self, file_path, document_id=None, metadata=None, progress_callback=None):
-        """
-        Xử lý file PDF, chia thành chunks và tạo embeddings
+        """Process a PDF file and create vector embeddings
         
-        Args:
-            file_path (str): Đường dẫn tới file PDF
-            document_id (str, optional): ID của tài liệu, nếu không cung cấp sẽ tạo ID mới
-            metadata (dict, optional): Metadata bổ sung cho tài liệu
-            progress_callback (callable, optional): Callback function để cập nhật tiến độ
-            
-        Returns:
-            dict: Thông tin kết quả xử lý gồm document_id và số chunks đã xử lý
+        This method:
+        1. Extracts text from PDF using PyPDFLoader
+        2. Splits text into chunks using RecursiveCharacterTextSplitter
+        3. Creates embeddings using Google Gemini model
+        4. Stores embeddings in Pinecone
         """
+        logger.info(f"[{self.correlation_id}] Processing PDF: {file_path}")
+        
+        if self.mock_mode:
+            logger.info(f"[{self.correlation_id}] MOCK: Processing PDF {file_path}")
+            # Mock implementation - just return success
+            if progress_callback:
+                await progress_callback(None, document_id, "embedding_complete", 1.0, "Mock processing completed")
+            return {"success": True, "message": "PDF processed successfully"}
+        
         try:
-            # Khởi tạo kết nối Pinecone nếu chưa có
-            if not self.pinecone_index:
-                if not self._init_pinecone_connection():
-                    return {"success": False, "error": "Không thể kết nối đến Pinecone"}
+            # Initialize metadata if not provided
+            if metadata is None:
+                metadata = {}
             
-            # Tạo document_id nếu không có
-            if not document_id:
+            # Ensure document_id is included
+            if document_id is None:
                 document_id = str(uuid.uuid4())
             
-            # Đọc file PDF bằng PyPDFLoader
-            logger.info(f"Đang đọc file PDF: {file_path}")
+            # Add document_id to metadata
+            metadata["document_id"] = document_id
+            
+            # The namespace to use might be in vdb-X format if vector_db_id provided
+            actual_namespace = f"vdb-{self.vector_db_id}" if self.vector_db_id else self.namespace
+            
+            # 1. Extract text from PDF
+            logger.info(f"[{self.correlation_id}] Extracting text from PDF: {file_path}")
             if progress_callback:
-                await progress_callback("pdf_loading", 0.5, "Loading PDF file")
+                await progress_callback(None, document_id, "text_extraction", 0.2, "Extracting text from PDF")
                 
             loader = PyPDFLoader(file_path)
-            pages = loader.load()
+            documents = loader.load()
+            total_text_length = sum(len(doc.page_content) for doc in documents)
             
-            # Trích xuất và nối text từ tất cả các trang
-            all_text = ""
-            for page in pages:
-                all_text += page.page_content + "\n"
+            logger.info(f"[{self.correlation_id}] Extracted {len(documents)} pages, total text length: {total_text_length}")
             
+            # 2. Split text into chunks
             if progress_callback:
-                await progress_callback("text_extraction", 0.6, "Extracted text from PDF")
-                
-            # Chia văn bản thành các chunk
-            text_splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=300)
-            chunks = text_splitter.split_text(all_text)
+                await progress_callback(None, document_id, "chunking", 0.4, "Splitting text into chunks")
+            
+            text_splitter = RecursiveCharacterTextSplitter(
+                chunk_size=1000,
+                chunk_overlap=100,
+                length_function=len,
+                separators=["\n\n", "\n", " ", ""]
+            )
+            
+            chunks = text_splitter.split_documents(documents)
             
-            logger.info(f"Đã chia file PDF thành {len(chunks)} chunks")
+            logger.info(f"[{self.correlation_id}] Split into {len(chunks)} chunks")
+            
+            # 3. Create embeddings
             if progress_callback:
-                await progress_callback("chunking", 0.7, f"Split document into {len(chunks)} chunks")
-            
-            # Xử lý embedding cho từng chunk và upsert lên Pinecone
-            vectors = []
-            for i, chunk in enumerate(chunks):
-                # Cập nhật tiến độ embedding
-                if progress_callback and i % 5 == 0:  # Cập nhật sau mỗi 5 chunks để tránh quá nhiều thông báo
-                    embedding_progress = 0.7 + (0.3 * (i / len(chunks)))
-                    await progress_callback("embedding", embedding_progress, f"Processing chunk {i+1}/{len(chunks)}")
+                await progress_callback(None, document_id, "embedding", 0.6, "Creating embeddings")
+            
+            # Initialize Google Gemini for embeddings
+            if not self.google_api_key:
+                raise ValueError("Google API key not found in environment variables")
+            
+            genai.configure(api_key=self.google_api_key)
+            
+            # First, get the expected dimensions from Pinecone
+            logger.info(f"[{self.correlation_id}] Checking Pinecone index dimensions")
+            if not self.pinecone_index:
+                self.pinecone_index = PineconeConnectionManager.get_index(self.api_key, self.index_name)
+            
+            stats = self.pinecone_index.describe_index_stats()
+            pinecone_dimension = stats.dimension
+            logger.info(f"[{self.correlation_id}] Pinecone index dimension: {pinecone_dimension}")
+            
+            # Create embedding model
+            embedding_model = GoogleGenerativeAIEmbeddings(
+                model="models/embedding-001",
+                google_api_key=self.google_api_key,
+                task_type="retrieval_document"  # Use document embedding mode for longer text
+            )
+            
+            # Get a sample embedding to check dimensions
+            sample_embedding = embedding_model.embed_query("test")
+            embedding_dimension = len(sample_embedding)
+            
+            logger.info(f"[{self.correlation_id}] Generated embeddings with dimension: {embedding_dimension}")
+            
+            # Dimension handling - if mismatch, we handle it appropriately
+            if embedding_dimension != pinecone_dimension:
+                logger.warning(f"[{self.correlation_id}] Embedding dimension mismatch: got {embedding_dimension}, need {pinecone_dimension}")
                 
-                # Tạo vector embedding cho từng chunk
-                vector = embeddings_model.embed_query(chunk)
+                if embedding_dimension < pinecone_dimension:
+                    # For upscaling from 768 to 1536: duplicate each value and scale appropriately
+                    # This is one approach to handle dimension mismatches while preserving semantic information
+                    logger.info(f"[{self.correlation_id}] Using duplication strategy to upscale from {embedding_dimension} to {pinecone_dimension}")
+                    
+                    if embedding_dimension * 2 == pinecone_dimension:
+                        # Perfect doubling (768 -> 1536)
+                        def adjust_embedding(embedding):
+                            # Duplicate each value to double the dimension
+                            return [val for val in embedding for _ in range(2)]
+                    else:
+                        # Generic padding with zeros
+                        pad_size = pinecone_dimension - embedding_dimension
+                        def adjust_embedding(embedding):
+                            return embedding + [0.0] * pad_size
+                else:
+                    # Truncation strategy - take first pinecone_dimension values
+                    logger.info(f"[{self.correlation_id}] Will truncate embeddings from {embedding_dimension} to {pinecone_dimension}")
+                    
+                    def adjust_embedding(embedding):
+                        return embedding[:pinecone_dimension]
+            else:
+                # No adjustment needed
+                def adjust_embedding(embedding):
+                    return embedding
+            
+            # Process in batches to avoid memory issues
+            batch_size = 10
+            vectors_to_upsert = []
+            
+            for i in range(0, len(chunks), batch_size):
+                batch = chunks[i:i+batch_size]
                 
-                # Chuẩn bị metadata cho vector
-                vector_metadata = {
-                    "document_id": document_id,
-                    "chunk_index": i,
-                    "text": chunk
-                }
+                # Extract text content
+                texts = [chunk.page_content for chunk in batch]
                 
-                # Thêm metadata bổ sung nếu có
-                if metadata:
-                    for key, value in metadata.items():
-                        if key not in vector_metadata:
-                            vector_metadata[key] = value
+                # Create embeddings for batch
+                embeddings = embedding_model.embed_documents(texts)
                 
-                # Thêm vector vào danh sách để upsert
-                vectors.append({
-                    "id": f"{document_id}_{i}",
-                    "values": vector,
-                    "metadata": vector_metadata
-                })
+                # Prepare vectors for Pinecone
+                for j, (chunk, embedding) in enumerate(zip(batch, embeddings)):
+                    # Adjust embedding dimensions if needed
+                    adjusted_embedding = adjust_embedding(embedding)
+                    
+                    # Verify dimensions are correct
+                    if len(adjusted_embedding) != pinecone_dimension:
+                        raise ValueError(f"Dimension mismatch after adjustment: got {len(adjusted_embedding)}, expected {pinecone_dimension}")
+                    
+                    # Create metadata for this chunk
+                    chunk_metadata = {
+                        "document_id": document_id,
+                        "page": chunk.metadata.get("page", 0),
+                        "chunk_id": f"{document_id}-chunk-{i+j}",
+                        "text": chunk.page_content[:1000],  # Store first 1000 chars of text
+                        **metadata  # Include original metadata
+                    }
+                    
+                    # Create vector record
+                    vector = {
+                        "id": f"{document_id}-{i+j}",
+                        "values": adjusted_embedding,
+                        "metadata": chunk_metadata
+                    }
+                    
+                    vectors_to_upsert.append(vector)
                 
-                # Upsert mỗi 100 vectors để tránh quá lớn
-                if len(vectors) >= 100:
-                    await self._upsert_vectors(vectors)
-                    vectors = []
+                logger.info(f"[{self.correlation_id}] Processed batch {i//batch_size + 1}/{(len(chunks)-1)//batch_size + 1}")
+            
+            # 4. Store embeddings in Pinecone
+            if progress_callback:
+                await progress_callback(None, document_id, "storing", 0.8, f"Storing {len(vectors_to_upsert)} vectors in Pinecone")
+            
+            logger.info(f"[{self.correlation_id}] Upserting {len(vectors_to_upsert)} vectors to Pinecone index {self.index_name}, namespace {actual_namespace}")
             
-            # Upsert các vectors còn lại
-            if vectors:
-                await self._upsert_vectors(vectors)
+            # Use PineconeConnectionManager for better error handling
+            result = PineconeConnectionManager.upsert_vectors_with_validation(
+                self.pinecone_index,
+                vectors_to_upsert,
+                namespace=actual_namespace
+            )
             
-            logger.info(f"Đã embedding và lưu {len(chunks)} chunks từ PDF với document_id: {document_id}")
+            logger.info(f"[{self.correlation_id}] Successfully upserted {result.get('upserted_count', 0)} vectors to Pinecone")
             
-            # Final progress update
             if progress_callback:
-                await progress_callback("completed", 1.0, "PDF processing complete")
+                await progress_callback(None, document_id, "embedding_complete", 1.0, "Processing completed")
             
+            # Return success with stats
             return {
                 "success": True,
                 "document_id": document_id,
                 "chunks_processed": len(chunks),
-                "total_text_length": len(all_text)
+                "total_text_length": total_text_length,
+                "vectors_created": len(vectors_to_upsert),
+                "vectors_upserted": result.get('upserted_count', 0),
+                "message": "PDF processed successfully"
             }
+        except Exception as e:
+            logger.error(f"[{self.correlation_id}] Error processing PDF: {str(e)}")
+            return {
+                "success": False,
+                "error": f"Error processing PDF: {str(e)}"
+            }
+    
+    async def list_namespaces(self):
+        """List all namespaces in the Pinecone index"""
+        if self.mock_mode:
+            logger.info(f"[{self.correlation_id}] MOCK: Listing namespaces")
+            return {"success": True, "namespaces": ["test"]}
+        
+        try:
+            if not self.pinecone_index:
+                self.pinecone_index = PineconeConnectionManager.get_index(self.api_key, self.index_name)
+            
+            # Get index stats which includes namespaces
+            stats = self.pinecone_index.describe_index_stats()
+            namespaces = list(stats.get("namespaces", {}).keys())
             
+            return {
+                "success": True,
+                "namespaces": namespaces
+            }
         except Exception as e:
-            logger.error(f"Lỗi khi xử lý PDF: {str(e)}")
-            if progress_callback:
-                await progress_callback("error", 0, f"Error processing PDF: {str(e)}")
+            logger.error(f"[{self.correlation_id}] Error listing namespaces: {str(e)}")
             return {
                 "success": False,
-                "error": str(e)
+                "error": f"Error listing namespaces: {str(e)}"
             }
     
-    async def _upsert_vectors(self, vectors):
-        """Upsert vectors vào Pinecone"""
+    async def delete_namespace(self):
+        """Delete all vectors in a namespace"""
+        if self.mock_mode:
+            logger.info(f"[{self.correlation_id}] MOCK: Deleting namespace '{self.namespace}'")
+            return {
+                "success": True,
+                "namespace": self.namespace,
+                "deleted_count": 100,
+                "message": f"Successfully deleted namespace '{self.namespace}'"
+            }
+
         try:
-            if not vectors:
-                return
+            if not self.pinecone_index:
+                self.pinecone_index = PineconeConnectionManager.get_index(self.api_key, self.index_name)
                 
-            result = self.pinecone_index.upsert(
-                vectors=vectors,
-                namespace=self.namespace
-            )
+            logger.info(f"[{self.correlation_id}] Deleting namespace '{self.namespace}' from index '{self.index_name}'")
+            
+            # Check if namespace exists
+            stats = self.pinecone_index.describe_index_stats()
+            namespaces = stats.get("namespaces", {})
             
-            logger.info(f"Đã upsert {len(vectors)} vectors vào Pinecone")
-            return result
+            if self.namespace in namespaces:
+                vector_count = namespaces[self.namespace].get("vector_count", 0)
+                # Delete all vectors in namespace
+                self.pinecone_index.delete(delete_all=True, namespace=self.namespace)
+                return {
+                    "success": True,
+                    "namespace": self.namespace,
+                    "deleted_count": vector_count,
+                    "message": f"Successfully deleted namespace '{self.namespace}' with {vector_count} vectors"
+                }
+            else:
+                return {
+                    "success": True,
+                    "namespace": self.namespace,
+                    "deleted_count": 0,
+                    "message": f"Namespace '{self.namespace}' does not exist - nothing to delete"
+                }
         except Exception as e:
-            logger.error(f"Lỗi khi upsert vectors: {str(e)}")
-            raise
+            logger.error(f"[{self.correlation_id}] Error deleting namespace: {str(e)}")
+            return {
+                "success": False,
+                "namespace": self.namespace,
+                "error": f"Error deleting namespace: {str(e)}"
+            }
     
-    async def delete_namespace(self):
-        """
-        Xóa toàn bộ vectors trong namespace hiện tại (tương đương xoá namespace).
-        """
-        # Khởi tạo kết nối nếu cần
-        if not self.pinecone_index and not self._init_pinecone_connection():
-            return {"success": False, "error": "Không thể kết nối đến Pinecone"}
+    async def delete_document(self, document_id):
+        """Delete vectors associated with a specific document ID"""
+        logger.info(f"[{self.correlation_id}] Deleting vectors for document '{document_id}' from namespace '{self.namespace}'")
+
+        if self.mock_mode:
+            logger.info(f"[{self.correlation_id}] MOCK: Deleting document vectors for '{document_id}'")
+            # In mock mode, simulate deleting 10 vectors
+            return {
+                "success": True,
+                "document_id": document_id,
+                "namespace": self.namespace,
+                "deleted_count": 10,
+                "message": f"Successfully deleted vectors for document '{document_id}' from namespace '{self.namespace}'"
+            }
 
         try:
-            # delete_all=True sẽ xóa toàn bộ vectors trong namespace
+            if not self.pinecone_index:
+                self.pinecone_index = PineconeConnectionManager.get_index(self.api_key, self.index_name)
+                
+            # Use metadata filtering to find vectors with matching document_id
+            # The specific namespace to use might be vdb-X format if vector_db_id provided
+            actual_namespace = f"vdb-{self.vector_db_id}" if self.vector_db_id else self.namespace
+            
+            # Search for vectors with this document ID
+            results = self.pinecone_index.query(
+                vector=[0] * 1536,  # Dummy vector, we only care about metadata filter
+                top_k=1,
+                include_metadata=True,
+                filter={"document_id": document_id},
+                namespace=actual_namespace
+            )
+            
+            # If no vectors found, return success with warning
+            if len(results.get("matches", [])) == 0:
+                logger.warning(f"[{self.correlation_id}] No vectors found for document '{document_id}' in namespace '{actual_namespace}'")
+                return {
+                    "success": True,
+                    "document_id": document_id,
+                    "namespace": actual_namespace,
+                    "deleted_count": 0,
+                    "warning": f"No vectors found for document '{document_id}' in namespace '{actual_namespace}'",
+                    "message": f"Successfully deleted 0 vectors for document '{document_id}' from namespace '{actual_namespace}'"
+                }
+            
+            # Delete vectors by filter
             result = self.pinecone_index.delete(
-                delete_all=True,
-                namespace=self.namespace
+                filter={"document_id": document_id},
+                namespace=actual_namespace
             )
-            logger.info(f"Đã xóa namespace '{self.namespace}' (tất cả vectors).")
-            return {"success": True, "detail": result}
+            
+            # Get delete count from result
+            deleted_count = result.get("deleted_count", 0)
+            
+            return {
+                "success": True,
+                "document_id": document_id,
+                "namespace": actual_namespace,
+                "deleted_count": deleted_count,
+                "message": f"Successfully deleted {deleted_count} vectors for document '{document_id}' from namespace '{actual_namespace}'"
+            }
         except Exception as e:
-            logger.error(f"Lỗi khi xóa namespace '{self.namespace}': {e}")
-            return {"success": False, "error": str(e)}
+            logger.error(f"[{self.correlation_id}] Error deleting document vectors: {str(e)}")
+            return {
+                "success": False,
+                "document_id": document_id,
+                "error": f"Error deleting document vectors: {str(e)}"
+            }
     
     async def list_documents(self):
-        """Lấy danh sách tất cả document_id từ Pinecone"""
+        """List all documents in the Pinecone index"""
+        if self.mock_mode:
+            logger.info(f"[{self.correlation_id}] MOCK: Listing documents in namespace '{self.namespace}'")
+            return {
+                "success": True,
+                "namespace": self.namespace,
+                "documents": [
+                    {"id": "doc1", "title": "Sample Document 1"},
+                    {"id": "doc2", "title": "Sample Document 2"}
+                ]
+            }
+            
         try:
-            # Khởi tạo kết nối Pinecone nếu chưa có
             if not self.pinecone_index:
-                if not self._init_pinecone_connection():
-                    return {"success": False, "error": "Không thể kết nối đến Pinecone"}
+                self.pinecone_index = PineconeConnectionManager.get_index(self.api_key, self.index_name)
+                
+            # The namespace to use might be in vdb-X format if vector_db_id provided
+            actual_namespace = f"vdb-{self.vector_db_id}" if self.vector_db_id else self.namespace
             
-            # Lấy thông tin index
+            # Get index stats
             stats = self.pinecone_index.describe_index_stats()
+            namespaces = stats.get("namespaces", {})
+            total_vectors = namespaces.get(actual_namespace, {}).get("vector_count", 0)
             
-            # Thực hiện truy vấn để lấy danh sách tất cả document_id duy nhất
-            # Phương pháp này có thể không hiệu quả với dataset lớn, nhưng là cách đơn giản nhất
-            # Trong thực tế, nên lưu danh sách document_id trong một database riêng
+            # Query unique document IDs
+            # Use a sparse vector with top_k=0 to just get metadata stats
+            # This is more efficient than retrieving actual vectors
+            results = self.pinecone_index.query(
+                vector=[0] * 1536,  # Dummy vector for metadata-only query
+                top_k=100,  # Limit to 100 results
+                include_metadata=True,
+                namespace=actual_namespace
+            )
+            
+            # Extract unique document IDs from metadata
+            document_map = {}
+            matches = results.get("matches", [])
+            
+            for match in matches:
+                metadata = match.get("metadata", {})
+                doc_id = metadata.get("document_id")
+                
+                if doc_id and doc_id not in document_map:
+                    document_map[doc_id] = {
+                        "id": doc_id,
+                        "title": metadata.get("title", "Unknown"),
+                        "chunks": 1
+                    }
+                elif doc_id:
+                    document_map[doc_id]["chunks"] += 1
+            
+            documents = list(document_map.values())
             
             return {
                 "success": True,
-                "total_vectors": stats.get('total_vector_count', 0),
-                "namespace": self.namespace,
-                "index_name": self.index_name
+                "namespace": actual_namespace,
+                "index_name": self.index_name,
+                "total_vectors": total_vectors,
+                "documents": documents
             }
         except Exception as e:
-            logger.error(f"Lỗi khi lấy danh sách documents: {str(e)}")
+            logger.error(f"[{self.correlation_id}] Error listing documents: {str(e)}")
             return {
                 "success": False,
-                "error": str(e)
-            } 
+                "error": f"Error listing documents: {str(e)}"
+            } 
+

app/utils/pinecone_fix.py

@@ -0,0 +1,194 @@
+"""
+Improved Pinecone connection handling with dimension validation.
+This module provides more robust connection and error handling for Pinecone operations.
+"""
+import logging
+import time
+from typing import Optional, Dict, Any, Tuple, List
+import pinecone
+from pinecone import Pinecone, ServerlessSpec, PodSpec
+
+logger = logging.getLogger(__name__)
+
+# Default retry settings
+DEFAULT_MAX_RETRIES = 3
+DEFAULT_RETRY_DELAY = 2
+
+class PineconeConnectionManager:
+    """
+    Manages Pinecone connections with enhanced error handling and dimension validation.
+    
+    This class centralizes Pinecone connection logic, providing:
+    - Connection pooling/reuse
+    - Automatic retries with exponential backoff
+    - Dimension validation before operations
+    - Detailed error logging for better debugging
+    """
+    
+    # Class-level cache of Pinecone clients
+    _clients = {}
+    
+    @classmethod
+    def get_client(cls, api_key: str) -> Pinecone:
+        """
+        Returns a Pinecone client for the given API key, creating one if needed.
+        
+        Args:
+            api_key: Pinecone API key
+            
+        Returns:
+            Initialized Pinecone client
+        """
+        if not api_key:
+            raise ValueError("Pinecone API key cannot be empty")
+            
+        # Return cached client if it exists
+        if api_key in cls._clients:
+            return cls._clients[api_key]
+            
+        # Log client creation (but hide full API key)
+        key_prefix = api_key[:4] + "..." if len(api_key) > 4 else "invalid"
+        logger.info(f"Creating new Pinecone client with API key (first 4 chars: {key_prefix}...)")
+        
+        try:
+            # Initialize Pinecone client
+            client = Pinecone(api_key=api_key)
+            cls._clients[api_key] = client
+            logger.info("Pinecone client created successfully")
+            return client
+        except Exception as e:
+            logger.error(f"Failed to create Pinecone client: {str(e)}")
+            raise RuntimeError(f"Pinecone client initialization failed: {str(e)}") from e
+    
+    @classmethod
+    def get_index(cls, 
+                  api_key: str, 
+                  index_name: str, 
+                  max_retries: int = DEFAULT_MAX_RETRIES) -> Any:
+        """
+        Get a Pinecone index with retry logic.
+        
+        Args:
+            api_key: Pinecone API key
+            index_name: Name of the index to connect to
+            max_retries: Maximum number of retry attempts
+            
+        Returns:
+            Pinecone index
+        """
+        client = cls.get_client(api_key)
+        
+        # Retry logic for connection issues
+        for attempt in range(max_retries):
+            try:
+                index = client.Index(index_name)
+                # Test the connection
+                _ = index.describe_index_stats()
+                logger.info(f"Connected to Pinecone index: {index_name}")
+                return index
+            except Exception as e:
+                if attempt < max_retries - 1:
+                    wait_time = DEFAULT_RETRY_DELAY * (2 ** attempt)  # Exponential backoff
+                    logger.warning(f"Pinecone connection attempt {attempt+1} failed: {e}. Retrying in {wait_time}s...")
+                    time.sleep(wait_time)
+                else:
+                    logger.error(f"Failed to connect to Pinecone index after {max_retries} attempts: {e}")
+                    raise RuntimeError(f"Pinecone index connection failed: {str(e)}") from e
+    
+    @classmethod
+    def validate_dimensions(cls, 
+                            index: Any, 
+                            vector_dimensions: int) -> Tuple[bool, Optional[str]]:
+        """
+        Validate that the vector dimensions match the Pinecone index configuration.
+        
+        Args:
+            index: Pinecone index
+            vector_dimensions: Dimensions of the vectors to be uploaded
+            
+        Returns:
+            Tuple of (is_valid, error_message)
+        """
+        try:
+            # Get index stats
+            stats = index.describe_index_stats()
+            index_dimensions = stats.dimension
+            
+            if index_dimensions != vector_dimensions:
+                error_msg = (f"Vector dimensions mismatch: Your vectors have {vector_dimensions} dimensions, "
+                            f"but Pinecone index expects {index_dimensions} dimensions")
+                logger.error(error_msg)
+                return False, error_msg
+                
+            return True, None
+        except Exception as e:
+            error_msg = f"Failed to validate dimensions: {str(e)}"
+            logger.error(error_msg)
+            return False, error_msg
+    
+    @classmethod
+    def upsert_vectors_with_validation(cls,
+                                    index: Any,
+                                    vectors: List[Dict[str, Any]],
+                                    namespace: str = "",
+                                    batch_size: int = 100) -> Dict[str, Any]:
+        """
+        Upsert vectors with dimension validation and batching.
+        
+        Args:
+            index: Pinecone index
+            vectors: List of vectors to upsert, each with 'id', 'values', and optional 'metadata'
+            namespace: Namespace to upsert to
+            batch_size: Size of batches for upserting
+            
+        Returns:
+            Result of upsert operation
+        """
+        if not vectors:
+            return {"upserted_count": 0, "success": True}
+            
+        # Validate dimensions with the first vector
+        if "values" in vectors[0] and len(vectors[0]["values"]) > 0:
+            vector_dim = len(vectors[0]["values"])
+            is_valid, error_msg = cls.validate_dimensions(index, vector_dim)
+            
+            if not is_valid:
+                logger.error(f"Dimension validation failed: {error_msg}")
+                raise ValueError(f"Vector dimensions do not match Pinecone index configuration: {error_msg}")
+        
+        # Batch upsert
+        total_upserted = 0
+        for i in range(0, len(vectors), batch_size):
+            batch = vectors[i:i+batch_size]
+            try:
+                result = index.upsert(vectors=batch, namespace=namespace)
+                batch_upserted = result.get("upserted_count", len(batch))
+                total_upserted += batch_upserted
+                logger.info(f"Upserted batch {i//batch_size + 1}: {batch_upserted} vectors")
+            except Exception as e:
+                logger.error(f"Failed to upsert batch {i//batch_size + 1}: {str(e)}")
+                raise RuntimeError(f"Vector upsert failed: {str(e)}") from e
+                
+        return {"upserted_count": total_upserted, "success": True}
+
+# Simplified function to check connection
+def check_connection(api_key: str, index_name: str) -> bool:
+    """
+    Test Pinecone connection and validate index exists.
+    
+    Args:
+        api_key: Pinecone API key
+        index_name: Name of index to test
+        
+    Returns:
+        True if connection successful, False otherwise
+    """
+    try:
+        index = PineconeConnectionManager.get_index(api_key, index_name)
+        stats = index.describe_index_stats()
+        total_vectors = stats.total_vector_count
+        logger.info(f"Pinecone connection is working. Total vectors: {total_vectors}")
+        return True
+    except Exception as e:
+        logger.error(f"Pinecone connection failed: {str(e)}")
+        return False