Apps Layer: HTTP & WebSocket Handlers

Purpose & Architecture

The apps/ sub-module contains all REST endpoint handlers and WebSocket routes for the RAGFlow API. It serves as the HTTP-to-domain adapter layer:

• REST Handlers: One Python file per resource (e.g., document.py, dialog.py, knowledge_base.py). Each handler is a Quart async function decorated with route predicates.
• Blueprint Registration: Endpoints are organized into blueprints in __init__.py. Multiple blueprints (e.g., bp_document, bp_dialog) are registered with the main Quart app at server startup.
• Auth Middleware: All handlers are protected by JWT verification (via auth_required decorator) and tenant isolation checks (multi-tenancy enforcement at the auth middleware level).
• Request Validation: Quart's request object provides access to JSON body, query params, and URL path params. Handlers validate request shape and delegate business logic to domain services.
• Response Shaping: Handlers convert domain objects (from services) to JSON responses and appropriate HTTP status codes (200 OK, 201 Created, 400 Bad Request, 403 Forbidden, 404 Not Found, 500 Internal Error).
• WebSocket Support: Some handlers support WebSocket upgrades (e.g., for real-time chat progress streaming). WebSocket handlers manage connection lifecycle and push updates to clients.

Handler Patterns & Request Flow

Typical REST Handler Structure

Each REST handler follows this pattern:

@bp_document.route('/documents', methods=['GET'])
@auth_required
async def list_documents():
    # 1. Extract user identity from request context (auth_required sets it)
    tenant_id = g.tenant_id
    user_id = g.user_id
    
    # 2. Parse request parameters
    limit = request.args.get('limit', 10, type=int)
    offset = request.args.get('offset', 0, type=int)
    
    # 3. Call service layer
    service = DocumentService()
    docs = service.list_documents(tenant_id, limit, offset)
    
    # 4. Shape response
    return jsonify({
        'success': True,
        'data': [doc.to_dict() for doc in docs],
        'total': len(docs)
    }), 200

Authentication & Auth Middleware

The auth_required decorator (from common/auth.py) performs these checks before the handler is invoked:

1. Extract JWT: Read Authorization header, extract Bearer token.
2. Decode JWT: Verify signature using secret key. If invalid, raise 401 Unauthorized.
3. Load User: Query database for user by sub claim. If not found, raise 401.
4. Load Tenant: Fetch user's default tenant_id. If tenant doesn't exist or is inactive, raise 403 Forbidden.
5. Set Request Context: Store user_id, tenant_id, user_object in Quart's g (request-scoped storage).
6. Verify Password Hash: For sensitive operations (account deletion, password reset), request body may include password. Verify via bcrypt before proceeding.

The tenant_id enforcement happens at TWO levels:

• Auth middleware: Ensures user belongs to tenant
• Service layer: Filters queries and validates mutations by tenant_id

Request Validation Patterns

Most handlers validate request shape to prevent downstream errors:

@bp_document.route('/documents', methods=['POST'])
@auth_required
async def create_document():
    # Validate request JSON
    data = await request.json
    if not data.get('name'):
        return jsonify({'error': 'Field "name" is required'}), 400
    
    if not isinstance(data.get('name'), str):
        return jsonify({'error': 'Field "name" must be a string'}), 400
    
    max_len = 255
    if len(data['name']) > max_len:
        return jsonify({'error': f'Field "name" must be <= {max_len} chars'}), 400
    
    # Proceed with validated data
    service = DocumentService()
    doc = service.create_document(
        tenant_id=g.tenant_id,
        user_id=g.user_id,
        name=data['name'],
        description=data.get('description', '')
    )
    return jsonify(doc.to_dict()), 201

Blueprint Organization

Blueprints in apps/__init__.py are organized by resource type:

Error Handling & HTTP Status Code Mapping

The apps layer converts domain exceptions to HTTP status codes. Common patterns:

try:
    service = DocumentService()
    doc = service.get_document(doc_id, tenant_id=g.tenant_id)
except DocumentNotFound:
    return jsonify({'error': 'Document not found'}), 404
except TenantMismatch:
    return jsonify({'error': 'Access denied'}), 403
except ValidationError as e:
    return jsonify({'error': str(e)}), 400
except InternalError as e:
    # Log error details
    logger.exception(f"Internal error: {e}")
    return jsonify({'error': 'Internal server error'}), 500

WebSocket Handlers & Real-Time Updates

Some handlers support WebSocket upgrade to stream real-time updates:

@bp_document.websocket('/documents/{doc_id}/progress')
@auth_required
async def stream_document_progress(doc_id):
    ws = websocket.receive()
    tenant_id = g.tenant_id
    
    try:
        doc = DocumentService().get_document(doc_id, tenant_id)
    except DocumentNotFound:
        return jsonify({'error': 'Document not found'}), 404
    
    # Subscribe to task progress updates
    task_id = doc.task_id
    
    # Stream progress from Redis every 500ms
    while True:
        progress = redis_client.get(f'task:{task_id}:progress')
        await websocket.send(json.dumps({
            'task_id': task_id,
            'progress': json.loads(progress)
        }))
        await asyncio.sleep(0.5)

Multi-Tenancy Enforcement in Apps Layer

The apps layer enforces tenant isolation at the handler level before delegating to services:

• Auth middleware: All @auth_required handlers have access to g.tenant_id, which is pre-verified to belong to the authenticated user.
• Explicit tenant_id passing: Handlers MUST pass tenant_id=g.tenant_id to all service calls. Never trust request parameters for tenant_id; use the auth middleware value.
• URL path isolation: Some routes are tenant-aware (e.g., /tenants/{tenant_id}/documents). Always validate that the URL path tenant_id matches g.tenant_id.

Key Files & Modules

• apps/__init__.py — Blueprint registration, Quart app setup
• apps/document.py — Document CRUD and upload handlers
• apps/dialog.py — Conversation/dialog handlers
• apps/knowledge_base.py — Knowledge base management handlers
• apps/chunk.py — Document chunk queries and reranking
• apps/task.py — Background task status and cancellation
• apps/llm.py — LLM model configuration
• common/auth.py — auth_required decorator, JWT decode, password hashing (bcrypt)
• common/exceptions.py — Custom exception classes (DocumentNotFound, TenantMismatch, ValidationError, etc.)

Design Decisions & Trade-offs

The apps layer embodies several key design decisions:

Performance Considerations

The apps layer is designed to be lightweight and fast:

• Auth latency: Each request incurs JWT decode (~1ms) and user/tenant DB query (~3–5ms). Consider caching user objects in g or Redis for repeated requests.
• Validation overhead: Request validation is O(1) for basic checks (field existence, type). Custom validators (e.g., name length, regex) add negligible overhead.
• Response serialization: Converting domain objects to JSON is typically <1ms. Use incremental JSON streaming for large responses (e.g., paginated lists).
• Blueprint registration: Blueprint registration happens at server startup, not per-request. Negligible runtime cost.

Security Implications

• JWT validation: All tokens must be valid and non-expired. Secret key must be strong and protected.
• Password hashing: Passwords are hashed with bcrypt before transmission. Verify hash on sensitive operations (account deletion, password reset).
• Tenant isolation: Apps layer enforces tenant_id checks before service calls. Missing checks = data leak.
• CORS & CSRF: API endpoints should enforce CORS headers to prevent cross-origin attacks. CSRF tokens not needed for stateless JWT auth.
• Rate limiting: Consider per-user or per-IP rate limits on sensitive endpoints (login, password reset, document upload).

Future Enhancements

• API Versioning: Introduce versioned endpoints (e.g., /api/v2/documents) to support gradual schema evolution without breaking clients.
• OpenAPI/Swagger: Generate OpenAPI spec from handlers; serve Swagger UI for interactive exploration.
• Response Pagination: Standardize pagination across all list endpoints; support cursor-based pagination for large datasets.
• Webhook Callbacks: Allow external systems to subscribe to events (document upload complete, task failure) via webhook URLs.
• GraphQL Alternative: Consider GraphQL endpoint for flexible querying of related resources in a single request.