Architecture Overview¶

// Queue processing with fixed thread pools
pub struct OcrQueueService {
    pool: PgPool,
    workers: Vec<JoinHandle<()>>,
    shutdown: Arc<AtomicBool>,
}

impl OcrQueueService {
    pub async fn start_workers(&self) {
        // Fixed thread allocation:
        // - OCR runtime: 3 threads
        // - Background runtime: 2 threads  
        // - Database runtime: 2 threads
        let ocr_workers = 3;

        for worker_id in 0..ocr_workers {
            let pool = self.pool.clone();
            let shutdown = self.shutdown.clone();

            let handle = tokio::spawn(async move {
                while !shutdown.load(Ordering::Relaxed) {
                    if let Some(job) = fetch_next_job(&pool).await {
                        process_ocr_job(job, &pool).await;
                    } else {
                        tokio::time::sleep(Duration::from_secs(1)).await;
                    }
                }
            });

            self.workers.push(handle);
        }
    }
}

// Storage trait for multiple backends
#[async_trait]
pub trait StorageBackend: Send + Sync {
    async fn store(&self, key: &str, data: &[u8]) -> Result<()>;
    async fn retrieve(&self, key: &str) -> Result<Vec<u8>>;
    async fn delete(&self, key: &str) -> Result<()>;
    async fn exists(&self, key: &str) -> Result<bool>;
    async fn list(&self, prefix: &str) -> Result<Vec<String>>;
}

// Implementations
pub struct LocalStorage { base_path: PathBuf }
pub struct S3Storage { bucket: String, client: S3Client }
pub struct AzureStorage { container: String, client: BlobClient }

pub struct WebDavSync {
    client: WebDavClient,
    db: Arc<DbConnection>,
    progress: Arc<Mutex<SyncProgress>>,
}

impl WebDavSync {
    pub async fn smart_sync(&self) -> Result<SyncResult> {
        // 1. Fetch remote file list with ETags
        let remote_files = self.client.list_files().await?;

        // 2. Compare with local database
        let local_files = self.db.get_source_files().await?;

        // 3. Determine changes
        let changes = self.calculate_changes(&remote_files, &local_files);

        // 4. Process changes in batches
        for batch in changes.chunks(100) {
            self.process_batch(batch).await?;
            self.update_progress().await?;
        }

        // 5. Clean up deleted files
        self.process_deletions(&remote_files, &local_files).await?;

        Ok(SyncResult { 
            added: changes.added.len(),
            updated: changes.updated.len(),
            deleted: changes.deleted.len()
        })
    }
}

pub struct SourceScheduler {
    sources: Arc<RwLock<Vec<Source>>>,
    executor: Arc<ThreadPool>,
}

impl SourceScheduler {
    pub async fn run(&self) {
        loop {
            let now = Utc::now();
            let sources = self.sources.read().await;

            for source in sources.iter() {
                if source.should_sync(now) {
                    let source_clone = source.clone();
                    self.executor.spawn(async move {
                        match source_clone.sync().await {
                            Ok(result) => log::info!("Sync completed: {:?}", result),
                            Err(e) => log::error!("Sync failed: {}", e),
                        }
                    });
                }
            }

            tokio::time::sleep(Duration::from_secs(60)).await;
        }
    }
}

// Database connection pool configuration
let pool = PgPoolOptions::new()
    .max_connections(32)
    .min_connections(5)
    .connect_timeout(Duration::from_secs(5))
    .acquire_timeout(Duration::from_secs(10))
    .idle_timeout(Duration::from_secs(600))
    .max_lifetime(Duration::from_secs(1800))
    .connect(&database_url)
    .await?;

// Multi-level caching
pub struct CacheManager {
    l1_cache: Arc<DashMap<String, CachedItem>>, // In-memory
    l2_cache: Option<RedisClient>,               // Redis (optional)
}

impl CacheManager {
    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Option<T> {
        // Check L1 cache
        if let Some(item) = self.l1_cache.get(key) {
            if !item.is_expired() {
                return Some(item.value.clone());
            }
        }

        // Check L2 cache
        if let Some(redis) = &self.l2_cache {
            if let Ok(value) = redis.get(key).await {
                self.l1_cache.insert(key.to_string(), value.clone());
                return Some(value);
            }
        }

        None
    }
}

// Batch document processing
pub async fn batch_process_documents(
    documents: Vec<Document>,
    batch_size: usize,
) -> Result<Vec<ProcessResult>> {
    let semaphore = Arc::new(Semaphore::new(batch_size));
    let mut tasks = Vec::new();

    for doc in documents {
        let permit = semaphore.clone().acquire_owned().await?;
        let task = tokio::spawn(async move {
            let result = process_document(doc).await;
            drop(permit);
            result
        });
        tasks.push(task);
    }

    let results = futures::future::join_all(tasks).await;
    Ok(results.into_iter().filter_map(Result::ok).collect())
}

// Role-based access control
#[derive(Debug, Clone, PartialEq)]
pub enum Role {
    Admin,    // Full system access
    Editor,   // Create, read, update, delete own documents
    Viewer,   // Read-only access to own documents
}

impl Role {
    pub fn can_upload(&self) -> bool {
        matches!(self, Role::Admin | Role::Editor)
    }

    pub fn can_delete(&self) -> bool {
        matches!(self, Role::Admin | Role::Editor)
    }

    pub fn can_manage_users(&self) -> bool {
        matches!(self, Role::Admin)
    }

    pub fn can_configure_system(&self) -> bool {
        matches!(self, Role::Admin)
    }
}

// Prometheus metrics
lazy_static! {
    static ref HTTP_REQUESTS: IntCounterVec = register_int_counter_vec!(
        "http_requests_total",
        "Total HTTP requests",
        &["method", "endpoint", "status"]
    ).unwrap();

    static ref OCR_PROCESSING_TIME: HistogramVec = register_histogram_vec!(
        "ocr_processing_duration_seconds",
        "OCR processing time",
        &["language", "status"]
    ).unwrap();

    static ref ACTIVE_USERS: IntGauge = register_int_gauge!(
        "active_users_total",
        "Number of active users"
    ).unwrap();
}

// OpenTelemetry integration
use opentelemetry::trace::Tracer;

pub async fn process_document_traced(doc: Document) -> Result<()> {
    let tracer = opentelemetry::global::tracer("readur");
    let span = tracer.start("process_document");
    let cx = Context::current_with_span(span);

    // Trace document loading
    let _load_span = tracer.start_with_context("load_document", &cx);
    let file_data = load_file(&doc.file_path).await?;

    // Trace OCR processing
    let _ocr_span = tracer.start_with_context("ocr_processing", &cx);
    let text = extract_text(&file_data).await?;

    // Trace database update
    let _db_span = tracer.start_with_context("update_database", &cx);
    update_document_content(&doc.id, &text).await?;

    Ok(())
}

// Unit test example
#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_document_processing() {
        let doc = create_test_document();
        let result = process_document(doc).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap().status, "completed");
    }

    #[tokio::test]
    async fn test_search_functionality() {
        let pool = create_test_pool().await;
        seed_test_data(&pool).await;

        let results = search_documents("test query", &pool).await;
        assert!(!results.is_empty());
    }
}