1use crate::parser::{parse_file, DatasourceDeclaration, ImportStatement, ParseError, PersistStatement, ParsedFile};
2use serde::Serialize;
3use std::collections::{HashMap, HashSet, VecDeque};
4use std::fs;
5use std::path::{Path, PathBuf};
6use thiserror::Error;
7
8#[derive(Error, Debug)]
9pub enum ResolveError {
10 #[error("Failed to read file {path}: {source}")]
11 IoError {
12 path: PathBuf,
13 source: std::io::Error,
14 },
15
16 #[error("Parse error in {path}: {source}")]
17 ParseError { path: PathBuf, source: ParseError },
18
19 #[error("Circular dependency detected: {cycle}")]
20 CircularDependency { cycle: String },
21
22 #[error("Import not found: {import_path} (resolved to {resolved_path})")]
23 ImportNotFound {
24 import_path: String,
25 resolved_path: PathBuf,
26 },
27}
28
29#[derive(Debug, Clone, Serialize)]
31pub struct ImportInfo {
32 pub raw_path: String,
33 pub alias: Option<String>,
34 pub resolved_path: Option<PathBuf>,
35 pub is_stdlib: bool,
36}
37
38impl From<&ImportStatement> for ImportInfo {
39 fn from(stmt: &ImportStatement) -> Self {
40 ImportInfo {
41 raw_path: stmt.raw_path.clone(),
42 alias: stmt.alias.clone(),
43 resolved_path: None,
44 is_stdlib: stmt.is_stdlib,
45 }
46 }
47}
48
49#[derive(Debug, Clone, Serialize)]
51pub struct DatasourceInfo {
52 pub name: String,
53}
54
55impl From<&DatasourceDeclaration> for DatasourceInfo {
56 fn from(ds: &DatasourceDeclaration) -> Self {
57 DatasourceInfo {
58 name: ds.name.clone(),
59 }
60 }
61}
62
63#[derive(Debug, Clone, Serialize)]
65pub struct PersistInfo {
66 pub mode: String,
67 pub target_datasource: String,
68}
69
70impl From<&PersistStatement> for PersistInfo {
71 fn from(ps: &PersistStatement) -> Self {
72 PersistInfo {
73 mode: ps.mode.to_string(),
74 target_datasource: ps.target_datasource.clone(),
75 }
76 }
77}
78
79#[derive(Debug, Clone, Serialize)]
81pub struct FileNode {
82 pub path: PathBuf,
84 pub relative_path: PathBuf,
86 pub imports: Vec<ImportInfo>,
88 pub datasources: Vec<DatasourceInfo>,
90 pub persists: Vec<PersistInfo>,
92 pub import_dependencies: Vec<PathBuf>,
94 pub updates_datasources: Vec<String>,
96 pub declares_datasources: Vec<String>,
98 pub depends_on_datasources: Vec<String>,
100}
101
102#[derive(Debug, Clone, Serialize)]
104pub struct DependencyGraph {
105 pub root: PathBuf,
107 pub order: Vec<PathBuf>,
109 pub files: HashMap<PathBuf, FileNode>,
111 pub datasource_declarations: HashMap<String, PathBuf>,
113 pub datasource_updaters: HashMap<String, Vec<PathBuf>>,
115 pub warnings: Vec<String>,
117}
118
119pub struct ImportResolver {
121 parsed_cache: HashMap<PathBuf, ParsedFile>,
123 warnings: Vec<String>,
125}
126
127impl ImportResolver {
128 pub fn new() -> Self {
129 Self {
130 parsed_cache: HashMap::new(),
131 warnings: Vec::new(),
132 }
133 }
134
135 pub fn resolve(&mut self, root_path: &Path) -> Result<DependencyGraph, ResolveError> {
137 let root_path = fs::canonicalize(root_path).map_err(|e| ResolveError::IoError {
138 path: root_path.to_path_buf(),
139 source: e,
140 })?;
141
142 let root_dir = root_path.parent().unwrap_or(Path::new("."));
143 let mut files: HashMap<PathBuf, FileNode> = HashMap::new();
144
145 let mut queue: VecDeque<PathBuf> = VecDeque::new();
147 let mut seen: HashSet<PathBuf> = HashSet::new();
148
149 queue.push_back(root_path.clone());
150 seen.insert(root_path.clone());
151
152 while let Some(current_path) = queue.pop_front() {
153 let parsed = self.parse_file(¤t_path)?;
154 let file_dir = current_path.parent().unwrap_or(Path::new("."));
155
156 let mut import_infos: Vec<ImportInfo> = Vec::new();
157 let mut import_dependencies: Vec<PathBuf> = Vec::new();
158
159 for import in &parsed.imports {
160 let mut info = ImportInfo::from(import);
161
162 if import.is_stdlib {
163 import_infos.push(info);
164 continue;
165 }
166
167 if let Some(resolved) = import.resolve(file_dir) {
168 if resolved.exists() {
169 let canonical =
170 fs::canonicalize(&resolved).map_err(|e| ResolveError::IoError {
171 path: resolved.clone(),
172 source: e,
173 })?;
174
175 info.resolved_path = Some(canonical.clone());
176 import_dependencies.push(canonical.clone());
177
178 if !seen.contains(&canonical) {
179 seen.insert(canonical.clone());
180 queue.push_back(canonical);
181 }
182 } else {
183 self.warnings.push(format!(
184 "Import '{}' in {} resolved to non-existent file: {}",
185 import.raw_path,
186 current_path.display(),
187 resolved.display()
188 ));
189 }
190 }
191
192 import_infos.push(info);
193 }
194
195 let datasource_infos: Vec<DatasourceInfo> =
196 parsed.datasources.iter().map(DatasourceInfo::from).collect();
197 let persist_infos: Vec<PersistInfo> =
198 parsed.persists.iter().map(PersistInfo::from).collect();
199
200 let declares_datasources: Vec<String> =
201 parsed.datasources.iter().map(|d| d.name.clone()).collect();
202 let updates_datasources: Vec<String> = parsed
203 .persists
204 .iter()
205 .map(|p| p.target_datasource.clone())
206 .collect();
207
208 let relative_path = pathdiff::diff_paths(¤t_path, root_dir)
209 .unwrap_or_else(|| current_path.clone());
210
211 files.insert(
212 current_path.clone(),
213 FileNode {
214 path: current_path,
215 relative_path,
216 imports: import_infos,
217 datasources: datasource_infos,
218 persists: persist_infos,
219 import_dependencies,
220 updates_datasources,
221 declares_datasources,
222 depends_on_datasources: Vec::new(), },
224 );
225 }
226
227 let mut datasource_declarations: HashMap<String, PathBuf> = HashMap::new();
229 let mut datasource_updaters: HashMap<String, Vec<PathBuf>> = HashMap::new();
230
231 for (path, node) in &files {
232 for ds_name in &node.declares_datasources {
233 if let Some(existing) = datasource_declarations.get(ds_name) {
234 self.warnings.push(format!(
235 "Datasource '{}' declared in multiple files: {} and {}",
236 ds_name,
237 existing.display(),
238 path.display()
239 ));
240 } else {
241 datasource_declarations.insert(ds_name.clone(), path.clone());
242 }
243 }
244
245 for ds_name in &node.updates_datasources {
246 datasource_updaters
247 .entry(ds_name.clone())
248 .or_insert_with(Vec::new)
249 .push(path.clone());
250 }
251 }
252
253 self.compute_datasource_dependencies(&mut files, &datasource_declarations);
255
256 let order =
258 self.topological_sort_with_datasources(&files, &datasource_declarations, &datasource_updaters)?;
259
260 Ok(DependencyGraph {
261 root: root_path,
262 order,
263 files,
264 datasource_declarations,
265 datasource_updaters,
266 warnings: self.warnings.clone(),
267 })
268 }
269
270 fn parse_file(&mut self, path: &Path) -> Result<ParsedFile, ResolveError> {
271 if let Some(cached) = self.parsed_cache.get(path) {
272 return Ok(cached.clone());
273 }
274
275 let content = fs::read_to_string(path).map_err(|e| ResolveError::IoError {
276 path: path.to_path_buf(),
277 source: e,
278 })?;
279
280 let parsed = parse_file(&content).map_err(|e| ResolveError::ParseError {
281 path: path.to_path_buf(),
282 source: e,
283 })?;
284
285 self.parsed_cache.insert(path.to_path_buf(), parsed.clone());
286 Ok(parsed)
287 }
288
289 fn compute_datasource_dependencies(
291 &self,
292 files: &mut HashMap<PathBuf, FileNode>,
293 _datasource_declarations: &HashMap<String, PathBuf>,
294 ) {
295 let paths: Vec<PathBuf> = files.keys().cloned().collect();
297
298 for path in paths {
299 let mut reachable_datasources: HashSet<String> = HashSet::new();
300 let mut visited: HashSet<PathBuf> = HashSet::new();
301 let mut stack: Vec<PathBuf> = vec![path.clone()];
302
303 while let Some(current) = stack.pop() {
304 if visited.contains(¤t) {
305 continue;
306 }
307 visited.insert(current.clone());
308
309 if let Some(node) = files.get(¤t) {
310 if current != path {
312 for ds in &node.declares_datasources {
313 reachable_datasources.insert(ds.clone());
314 }
315 }
316
317 for dep in &node.import_dependencies {
319 if !visited.contains(dep) {
320 stack.push(dep.clone());
321 }
322 }
323 }
324 }
325
326 if let Some(node) = files.get_mut(&path) {
327 node.depends_on_datasources = reachable_datasources.into_iter().collect();
328 }
329 }
330 }
331
332 fn topological_sort_with_datasources(
340 &self,
341 files: &HashMap<PathBuf, FileNode>,
342 datasource_declarations: &HashMap<String, PathBuf>,
343 datasource_updaters: &HashMap<String, Vec<PathBuf>>,
344 ) -> Result<Vec<PathBuf>, ResolveError> {
345 let mut edges: HashMap<PathBuf, HashSet<PathBuf>> = HashMap::new();
348
349 for path in files.keys() {
351 edges.insert(path.clone(), HashSet::new());
352 }
353
354 for (path, node) in files {
356 for dep in &node.import_dependencies {
358 if files.contains_key(dep) {
359 edges.get_mut(dep).unwrap().insert(path.clone());
360 }
361 }
362
363 for ds_name in &node.declares_datasources {
368 if let Some(updaters) = datasource_updaters.get(ds_name) {
369 for updater_path in updaters {
370 if updater_path != path && files.contains_key(updater_path) {
371 let updater_node = files.get(updater_path).unwrap();
373 let imports_declarer = updater_node.import_dependencies.contains(path);
374
375 if !imports_declarer {
377 edges.get_mut(updater_path).unwrap().insert(path.clone());
379 }
380 }
381 }
382 }
383 }
384
385 for ds_name in &node.depends_on_datasources {
388 if let Some(declaring_path) = datasource_declarations.get(ds_name) {
389 if declaring_path != path && files.contains_key(declaring_path) {
390 edges.get_mut(declaring_path).unwrap().insert(path.clone());
392 }
393 }
394 }
395 }
396
397 let mut in_degree: HashMap<PathBuf, usize> = HashMap::new();
399 for path in files.keys() {
400 in_degree.insert(path.clone(), 0);
401 }
402
403 for dependents in edges.values() {
404 for dep in dependents {
405 *in_degree.get_mut(dep).unwrap() += 1;
406 }
407 }
408
409 let mut queue: VecDeque<PathBuf> = VecDeque::new();
410 let mut result: Vec<PathBuf> = Vec::new();
411
412 for (path, °ree) in &in_degree {
414 if degree == 0 {
415 queue.push_back(path.clone());
416 }
417 }
418
419 while let Some(current) = queue.pop_front() {
420 result.push(current.clone());
421
422 if let Some(dependents) = edges.get(¤t) {
423 for dependent in dependents {
424 let degree = in_degree.get_mut(dependent).unwrap();
425 *degree -= 1;
426 if *degree == 0 {
427 queue.push_back(dependent.clone());
428 }
429 }
430 }
431 }
432
433 if result.len() != files.len() {
435 let remaining: Vec<_> = files
436 .keys()
437 .filter(|p| !result.contains(p))
438 .map(|p| p.display().to_string())
439 .collect();
440 return Err(ResolveError::CircularDependency {
441 cycle: remaining.join(" -> "),
442 });
443 }
444
445 Ok(result)
446 }
447}
448
449impl Default for ImportResolver {
450 fn default() -> Self {
451 Self::new()
452 }
453}
454
455#[cfg(test)]
456mod tests {
457 use super::*;
458 use std::fs;
459 use tempfile::TempDir;
460
461 fn create_test_file(dir: &Path, name: &str, content: &str) -> PathBuf {
462 let path = dir.join(name);
463 if let Some(parent) = path.parent() {
464 fs::create_dir_all(parent).unwrap();
465 }
466 fs::write(&path, content).unwrap();
467 path
468 }
469
470 #[test]
471 fn test_simple_resolution() {
472 let temp = TempDir::new().unwrap();
473 let root = temp.path();
474
475 create_test_file(root, "a.preql", "import b;");
476 create_test_file(root, "b.preql", "// no imports");
477
478 let a_path = root.join("a.preql");
479 let mut resolver = ImportResolver::new();
480 let graph = resolver.resolve(&a_path).unwrap();
481
482 assert_eq!(graph.order.len(), 2);
483 let b_idx = graph
485 .order
486 .iter()
487 .position(|p| p.ends_with("b.preql"))
488 .unwrap();
489 let a_idx = graph
490 .order
491 .iter()
492 .position(|p| p.ends_with("a.preql"))
493 .unwrap();
494 assert!(b_idx < a_idx, "b should come before a");
495 }
496
497 #[test]
498 fn test_datasource_declaration_ordering() {
499 let temp = TempDir::new().unwrap();
500 let root = temp.path();
501
502 create_test_file(root, "a.preql", "import b;");
507 create_test_file(
508 root,
509 "b.preql",
510 r#"
511 datasource orders (
512 id: key
513 )
514 address db.orders;
515 "#,
516 );
517 create_test_file(root, "c.preql", "persist orders;");
518
519 let a_path = root.join("a.preql");
521 let mut resolver = ImportResolver::new();
522 let graph = resolver.resolve(&a_path).unwrap();
523
524 assert!(graph.datasource_declarations.contains_key("orders"));
526
527 let b_idx = graph.order.iter().position(|p| p.ends_with("b.preql")).unwrap();
529 let a_idx = graph.order.iter().position(|p| p.ends_with("a.preql")).unwrap();
530 assert!(b_idx < a_idx, "b should come before a due to import");
531 }
532
533 #[test]
534 fn test_persist_before_declare() {
535 let temp = TempDir::new().unwrap();
536 let root = temp.path();
537
538 create_test_file(root, "updater.preql", "persist orders;");
543 create_test_file(
544 root,
545 "declarer.preql",
546 r#"
547 import updater;
548 datasource orders (
549 id: key
550 )
551 address db.orders;
552 "#,
553 );
554
555 let declarer_path = root.join("declarer.preql");
556 let mut resolver = ImportResolver::new();
557 let graph = resolver.resolve(&declarer_path).unwrap();
558
559 let updater_idx = graph
560 .order
561 .iter()
562 .position(|p| p.ends_with("updater.preql"))
563 .unwrap();
564 let declarer_idx = graph
565 .order
566 .iter()
567 .position(|p| p.ends_with("declarer.preql"))
568 .unwrap();
569
570 assert!(
571 updater_idx < declarer_idx,
572 "updater (persist) should come before declarer (datasource)"
573 );
574 }
575
576 #[test]
577 fn test_full_dependency_chain() {
578 let temp = TempDir::new().unwrap();
579 let root = temp.path();
580
581 create_test_file(
592 root,
593 "base.preql",
594 r#"
595 datasource orders (
596 id: key,
597 amount: metric
598 )
599 address db.orders;
600 "#,
601 );
602 create_test_file(
603 root,
604 "updater.preql",
605 r#"
606 persist orders where amount > 100;
607 "#,
608 );
609 create_test_file(
610 root,
611 "consumer.preql",
612 r#"
613 import base;
614 // uses orders datasource
615 "#,
616 );
617
618 create_test_file(
620 root,
621 "main.preql",
622 r#"
623 import updater;
624 import consumer;
625 "#,
626 );
627
628 let main_path = root.join("main.preql");
629 let mut resolver = ImportResolver::new();
630 let graph = resolver.resolve(&main_path).unwrap();
631
632 assert_eq!(graph.order.len(), 4);
633
634 let updater_idx = graph
635 .order
636 .iter()
637 .position(|p| p.ends_with("updater.preql"))
638 .unwrap();
639 let base_idx = graph
640 .order
641 .iter()
642 .position(|p| p.ends_with("base.preql"))
643 .unwrap();
644 let consumer_idx = graph
645 .order
646 .iter()
647 .position(|p| p.ends_with("consumer.preql"))
648 .unwrap();
649 let main_idx = graph
650 .order
651 .iter()
652 .position(|p| p.ends_with("main.preql"))
653 .unwrap();
654
655 assert!(
657 updater_idx < base_idx,
658 "updater should come before base: updater={}, base={}",
659 updater_idx,
660 base_idx
661 );
662
663 assert!(
665 base_idx < consumer_idx,
666 "base should come before consumer: base={}, consumer={}",
667 base_idx,
668 consumer_idx
669 );
670
671 assert!(
673 updater_idx < main_idx && base_idx < main_idx && consumer_idx < main_idx,
674 "main should come after all others"
675 );
676 }
677
678 #[test]
679 fn test_multiple_datasources() {
680 let temp = TempDir::new().unwrap();
681 let root = temp.path();
682
683 create_test_file(
684 root,
685 "models.preql",
686 r#"
687 datasource customers (
688 id: key
689 )
690 address db.customers;
691
692 datasource orders (
693 id: key,
694 customer_id
695 )
696 address db.orders;
697 "#,
698 );
699
700 let models_path = root.join("models.preql");
701 let mut resolver = ImportResolver::new();
702 let graph = resolver.resolve(&models_path).unwrap();
703
704 assert_eq!(graph.datasource_declarations.len(), 2);
705 assert!(graph.datasource_declarations.contains_key("customers"));
706 assert!(graph.datasource_declarations.contains_key("orders"));
707 }
708}