floe_core/config/
filesystem.rs

1use std::collections::HashMap;
2use std::path::{Path, PathBuf};
3
4use crate::config::{FilesystemDefinition, RootConfig};
5use crate::{ConfigError, FloeResult};
6
7#[derive(Debug, Clone)]
8pub struct ResolvedPath {
9    pub filesystem: String,
10    pub uri: String,
11    pub local_path: Option<PathBuf>,
12}
13
14pub struct FilesystemResolver {
15    config_dir: PathBuf,
16    default_name: String,
17    definitions: HashMap<String, FilesystemDefinition>,
18    has_config: bool,
19}
20
21impl FilesystemResolver {
22    pub fn new(config: &RootConfig, config_path: &Path) -> FloeResult<Self> {
23        let config_dir = config_path
24            .parent()
25            .unwrap_or_else(|| Path::new("."))
26            .to_path_buf();
27        if let Some(filesystems) = &config.filesystems {
28            let mut definitions = HashMap::new();
29            for definition in &filesystems.definitions {
30                if definitions
31                    .insert(definition.name.clone(), definition.clone())
32                    .is_some()
33                {
34                    return Err(Box::new(ConfigError(format!(
35                        "filesystems.definitions name={} is duplicated",
36                        definition.name
37                    ))));
38                }
39            }
40            let default_name = filesystems.default.clone().ok_or_else(|| {
41                Box::new(ConfigError("filesystems.default is required".to_string()))
42            })?;
43            if !definitions.contains_key(&default_name) {
44                return Err(Box::new(ConfigError(format!(
45                    "filesystems.default={} does not match any definition",
46                    default_name
47                ))));
48            }
49            Ok(Self {
50                config_dir,
51                default_name,
52                definitions,
53                has_config: true,
54            })
55        } else {
56            Ok(Self {
57                config_dir,
58                default_name: "local".to_string(),
59                definitions: HashMap::new(),
60                has_config: false,
61            })
62        }
63    }
64
65    pub fn resolve_path(
66        &self,
67        entity_name: &str,
68        field: &str,
69        filesystem_name: Option<&str>,
70        raw_path: &str,
71    ) -> FloeResult<ResolvedPath> {
72        let name = filesystem_name.unwrap_or(self.default_name.as_str());
73        if !self.has_config && name != "local" {
74            return Err(Box::new(ConfigError(format!(
75                "entity.name={} {field} references unknown filesystem {} (no filesystems block)",
76                entity_name, name
77            ))));
78        }
79
80        let definition = if self.has_config {
81            self.definitions.get(name).cloned().ok_or_else(|| {
82                Box::new(ConfigError(format!(
83                    "entity.name={} {field} references unknown filesystem {}",
84                    entity_name, name
85                )))
86            })?
87        } else {
88            FilesystemDefinition {
89                name: "local".to_string(),
90                fs_type: "local".to_string(),
91                bucket: None,
92                region: None,
93                prefix: None,
94            }
95        };
96
97        match definition.fs_type.as_str() {
98            "local" => {
99                let resolved = resolve_local_path(&self.config_dir, raw_path);
100                Ok(ResolvedPath {
101                    filesystem: name.to_string(),
102                    uri: local_uri(&resolved),
103                    local_path: Some(resolved),
104                })
105            }
106            "s3" => {
107                let uri = resolve_s3_uri(&definition, raw_path)?;
108                Ok(ResolvedPath {
109                    filesystem: name.to_string(),
110                    uri,
111                    local_path: None,
112                })
113            }
114            _ => Err(Box::new(ConfigError(format!(
115                "filesystem type {} is unsupported",
116                definition.fs_type
117            )))),
118        }
119    }
120
121    pub fn definition(&self, name: &str) -> Option<FilesystemDefinition> {
122        if self.has_config {
123            self.definitions.get(name).cloned()
124        } else if name == "local" {
125            Some(FilesystemDefinition {
126                name: "local".to_string(),
127                fs_type: "local".to_string(),
128                bucket: None,
129                region: None,
130                prefix: None,
131            })
132        } else {
133            None
134        }
135    }
136}
137
138pub fn resolve_local_path(config_dir: &Path, raw_path: &str) -> PathBuf {
139    let path = Path::new(raw_path);
140    if path.is_absolute() {
141        path.to_path_buf()
142    } else {
143        config_dir.join(path)
144    }
145}
146
147fn local_uri(path: &Path) -> String {
148    format!("local://{}", path.display())
149}
150
151fn resolve_s3_uri(definition: &FilesystemDefinition, raw_path: &str) -> FloeResult<String> {
152    let bucket = definition.bucket.as_ref().ok_or_else(|| {
153        Box::new(ConfigError(format!(
154            "filesystem {} requires bucket for type s3",
155            definition.name
156        )))
157    })?;
158    if let Some((bucket_in_path, key)) = parse_s3_uri(raw_path) {
159        if bucket_in_path != *bucket {
160            return Err(Box::new(ConfigError(format!(
161                "filesystem {} bucket mismatch: {}",
162                definition.name, bucket_in_path
163            ))));
164        }
165        return Ok(format_s3_uri(bucket, &key));
166    }
167
168    let key = join_s3_key(definition.prefix.as_deref().unwrap_or(""), raw_path);
169    Ok(format_s3_uri(bucket, &key))
170}
171
172fn parse_s3_uri(value: &str) -> Option<(String, String)> {
173    let stripped = value.strip_prefix("s3://")?;
174    let mut parts = stripped.splitn(2, '/');
175    let bucket = parts.next()?.to_string();
176    if bucket.is_empty() {
177        return None;
178    }
179    let key = parts.next().unwrap_or("").to_string();
180    Some((bucket, key))
181}
182
183fn join_s3_key(prefix: &str, raw_path: &str) -> String {
184    let prefix = prefix.trim_matches('/');
185    let trimmed = raw_path.trim_start_matches('/');
186    match (prefix.is_empty(), trimmed.is_empty()) {
187        (true, true) => String::new(),
188        (true, false) => trimmed.to_string(),
189        (false, true) => prefix.to_string(),
190        (false, false) => format!("{}/{}", prefix, trimmed),
191    }
192}
193
194fn format_s3_uri(bucket: &str, key: &str) -> String {
195    if key.is_empty() {
196        format!("s3://{}", bucket)
197    } else {
198        format!("s3://{}/{}", bucket, key)
199    }
200}
floe_core/config/filesystem.rs

floe_core/config/
filesystem.rs
