feat: Implement RQ-based PyMuPDF integration for async PDF processing

[priyankeshh]

feat: Implement RQ-based PyMuPDF integration for async PDF processing

🎯 Overview

This PR implements Phase 2 of the RQ-PyMuPDF integration plan, adding direct Redis Queue (RQ) communication between extralit-server and extralit-hf-space workers for asynchronous PDF processing with AGPL compliance.

🏗️ Architecture Changes

Before (HTTP-based)

extralit-server → HTTP → extralit-hf-space FastAPI → PyMuPDF

After (RQ-based)

extralit-server → RQ (pdf_queue) → extralit-hf-space worker → PyMuPDF

✨ Key Features

• 🚀 Async Processing: Non-blocking PDF extraction using Redis queues
• ⚖️ AGPL Compliance: Strict separation of Apache 2.0 and AGPL code via RQ workers
• 🔄 Fallback Mechanism: HTTP fallback when Redis is unavailable
• 📊 Smart Policies: File size limits and extraction configuration
• 🎛️ Configurable: Environment-driven RQ enable/disable

📁 Files Added

Core RQ Integration

• src/extralit_server/contexts/ocr/rq_client.py - Direct RQ client for pdf_queue
• src/extralit_server/jobs/pdf_extraction_jobs.py - PDF extraction orchestration with RQ polling

Enhanced Pipeline

• Updated src/extralit_server/jobs/document_jobs.py - Integrated RQ extraction into document processing
• Updated pyproject.toml - Added Redis dependency

🔧 Configuration

Environment Variables

# Enable/disable RQ processing
PYMUPDF_RQ_ENABLED=true
PYMUPDF_RQ_FALLBACK_HTTP=true

# Redis connection
REDIS_URL=redis://localhost:6379/0
PYMUPDF_EXTRACTION_QUEUE=pdf_queue

# Fallback HTTP service
PYMUPDF_SERVICE_URL=http://localhost:7860

🧪 Testing

Quick Test

# Start Redis
docker run -d -p 6379:6379 redis:alpine

# Start extralit-hf-space worker (separate repo)
cd extralit-hf-space
git checkout feat/rq-pymupdf-integration
python -m src.worker &

# Test RQ integration
cd extralit-server
python -c "
from src.extralit_server.contexts.ocr.rq_client import is_redis_available
print(f'Redis available: {is_redis_available()}')
"

Integration Testing

The RQ client includes comprehensive testing utilities for:

• ✅ Redis connectivity validation
• ✅ Job enqueueing and status monitoring
• ✅ HTTP fallback mechanism
• ✅ Error handling and timeout scenarios

🔄 Backward Compatibility

• ✅ Existing workflows unchanged - HTTP extraction still available as fallback
• ✅ Configuration-driven - RQ can be disabled via environment variables
• ✅ Graceful degradation - Automatic fallback to HTTP when Redis unavailable

📋 Implementation Details

Direct RQ Communication

• Uses string job reference: "src.jobs.extraction_jobs.extract_pdf_markdown_job"
• Connects to same pdf_queue as extralit-hf-space worker
• No HTTP dependency in RQ workflow (pure Redis communication)

Smart Extraction Policies

def should_extract_text(filename: str, file_metadata: dict) -> bool:
    """Smart extraction policies based on file type and size."""
    if not filename.lower().endswith('.pdf'):
        return False
    
    file_size = file_metadata.get('size', 0)
    max_size = 50 * 1024 * 1024  # 50MB limit
    return file_size <= max_size

Error Handling

• Timeout handling: Configurable job timeouts with polling
• Redis failures: Automatic HTTP fallback
• Job failures: Detailed error reporting and retry logic

🔗 Related Changes

This PR works in conjunction with:

• extralit-hf-space feat/rq-pymupdf-integration branch - Pure RQ worker implementation
• Maintains compatibility with existing PyMuPDF HTTP service

Branch: feat/rq-pymupdf-integration
Type: Feature
Breaking Changes: None
Dependencies: Redis server, extralit-hf-space RQ worker

[JonnyTran]

You can import from extralit_server.jobs.queues import REDIS_CONNECTION

from extralit_server.settings import settings

if settings.redis_use_cluster:
    REDIS_CONNECTION = RedisCluster.from_url(settings.redis_url)
else:
    REDIS_CONNECTION = redis.from_url(settings.redis_url)

[JonnyTran]

see comments

[JonnyTran]

I don't think we need this file, since it's already being called in start_pdf_workflow

[JonnyTran]

This pull request introduces a new job queue specifically for PDF OCR processing and updates the PDF workflow to utilize this new queue for text extraction jobs. The main goal is to better organize and route jobs related to PDF OCR, separating them from the default queue, and to prepare for future workflow enhancements.

Job queue improvements:

• Added a new PDF_OCR_QUEUE to queues.py for handling PDF OCR-related jobs, ensuring these tasks are managed separately from other job types.

Workflow enhancements:

• Updated the PDF workflow in pdf.py to enqueue PyMuPDF-based text extraction jobs to the new PDF_OCR_QUEUE, with dependencies and job configuration, instead of the default queue. [1] [2]
• Modified the workflow logic to store the new text extraction job ID in the workflow record, ensuring traceability and easier management of job dependencies.