orbok_workers/
extract.rs

1//! Extraction worker (RFC-005 §14): reads a queued file, runs the
2//! extractor, stores the output in the cache, and writes an
3//! extraction_record. On success, queues a Chunk job.
4
5use orbok_cache::{CacheService, EngineOptions, OrbokCacheNamespace};
6use orbok_core::ExtractionId;
7use orbok_core::{FileId, JobType, OrbokError, OrbokResult, now_iso8601};
8use orbok_db::Catalog;
9use orbok_db::repo::{FileRepository, IndexJobRepository, SourceRepository};
10use orbok_extract::{ExtractContext, ExtractOutput, ExtractorRegistry};
11use orbok_fs::{GuardedSource, PathGuard};
12use std::path::Path;
13
14/// Extraction worker instance, held for the duration of an index run.
15pub struct ExtractionWorker<'a> {
16    catalog: &'a Catalog,
17    cache: &'a CacheService,
18    registry: ExtractorRegistry,
19}
20
21impl<'a> ExtractionWorker<'a> {
22    pub fn new(catalog: &'a Catalog, cache: &'a CacheService) -> Self {
23        Self {
24            catalog,
25            cache,
26            registry: ExtractorRegistry::default(),
27        }
28    }
29
30    /// Run extraction for one file. Fails with a typed error on
31    /// unrecoverable cases; the worker coordinator converts failures to
32    /// a catalog record.
33    pub fn run(&self, file_id: &FileId) -> OrbokResult<()> {
34        let files = FileRepository::new(self.catalog);
35        let record = files.get_by_id(file_id)?.ok_or(OrbokError::FileNotFound)?;
36        let sources = SourceRepository::new(self.catalog);
37        let source = sources
38            .get(&record.source_id)?
39            .ok_or(OrbokError::SourceNotFound)?;
40
41        // Build a single-source path guard (RFC-003 §8 boundary).
42        let guard = PathGuard::new(vec![GuardedSource::from_record(&source)]);
43        let validated = guard.validate(Path::new(&record.canonical_path))?;
44
45        // Skip if cached extraction is still fresh (Appendix A §8).
46        let engine = self.cache.engine::<ExtractOutput>(
47            self.catalog,
48            &OrbokCacheNamespace::ExtractSegments,
49            EngineOptions::default(),
50        )?;
51        if CacheService::get_fresh(&engine, &validated)?.is_some() {
52            // Still fresh — queue the chunk job and return.
53            IndexJobRepository::new(self.catalog).enqueue(
54                JobType::Chunk,
55                Some(&record.source_id),
56                Some(file_id),
57            )?;
58            return Ok(());
59        }
60
61        // Run extractor with panic isolation and resource limits (RFC-044).
62        let output = self
63            .registry
64            .extract_safely(&validated, &ExtractContext::default())?;
65
66        // Cache the output (Appendix A §9.1).
67        CacheService::put(&engine, &validated, &output)?;
68
69        // Record in catalog.
70        let extraction_id = ExtractionId::generate();
71        let now = now_iso8601();
72        {
73            let conn = self.catalog.lock();
74            conn.execute(
75                "INSERT INTO extraction_records \
76                 (extraction_id, file_id, extractor_name, extractor_version, \
77                  normalization_version, source_content_hash, status, \
78                  extracted_char_count, extracted_byte_count, started_at, completed_at, \
79                  created_at, updated_at) \
80                 VALUES (?1,?2,?3,?4,?5,?6,'succeeded',?7,?8,?9,?9,?9,?9)",
81                rusqlite::params![
82                    extraction_id.as_str(),
83                    file_id.as_str(),
84                    output.extractor_name,
85                    output.extractor_version,
86                    output.normalization_version,
87                    record.content_hash,
88                    output.char_count as i64,
89                    output.char_count as i64,
90                    now,
91                ],
92            )
93            .map_err(|e| OrbokError::Database(e.to_string()))?;
94        }
95
96        // Queue chunk job.
97        IndexJobRepository::new(self.catalog).enqueue(
98            JobType::Chunk,
99            Some(&record.source_id),
100            Some(file_id),
101        )?;
102        Ok(())
103    }
104}
orbok_workers/extract.rs

orbok_workers/
extract.rs
