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::{ErrorCategory, FileId, JobType, OrbokError, OrbokResult, now_iso8601};
8use orbok_db::Catalog;
9use orbok_db::repo::{FileRepository, IndexJobRepository, SourceRepository};
10use orbok_extract::{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.
62        let output = self
63            .registry
64            .extract(&validated)
65            .map_err(|e| OrbokError::Extraction {
66                category: ErrorCategory::ParserError,
67                message: e.to_string(),
68            })?;
69
70        // Cache the output (Appendix A §9.1).
71        CacheService::put(&engine, &validated, &output)?;
72
73        // Record in catalog.
74        let extraction_id = ExtractionId::generate();
75        let now = now_iso8601();
76        {
77            let conn = self.catalog.lock();
78            conn.execute(
79                "INSERT INTO extraction_records \
80                 (extraction_id, file_id, extractor_name, extractor_version, \
81                  normalization_version, source_content_hash, status, \
82                  extracted_char_count, extracted_byte_count, started_at, completed_at, \
83                  created_at, updated_at) \
84                 VALUES (?1,?2,?3,?4,?5,?6,'succeeded',?7,?8,?9,?9,?9,?9)",
85                rusqlite::params![
86                    extraction_id.as_str(),
87                    file_id.as_str(),
88                    output.extractor_name,
89                    output.extractor_version,
90                    output.normalization_version,
91                    record.content_hash,
92                    output.char_count as i64,
93                    output.char_count as i64,
94                    now,
95                ],
96            )
97            .map_err(|e| OrbokError::Database(e.to_string()))?;
98        }
99
100        // Queue chunk job.
101        IndexJobRepository::new(self.catalog).enqueue(
102            JobType::Chunk,
103            Some(&record.source_id),
104            Some(file_id),
105        )?;
106        Ok(())
107    }
108}
orbok_workers/extract.rs

orbok_workers/
extract.rs
