flowsurface_data/chart/kline/indicators/
loader.rs

1use super::{IndicatorRegistration, language};
2use crate::data_path;
3use log::warn;
4use std::fs;
5use std::path::{Path, PathBuf};
6use std::sync::OnceLock;
7
8static INSTALLED_DIRECTORY: OnceLock<PathBuf> = OnceLock::new();
9
10/// Filesystem sources used to build the process-wide indicator catalog.
11///
12/// Bundled sources are managed by the data crate, while installed paths
13/// belong to the user or a development workflow.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct IndicatorCatalogPaths {
16    /// Overrides embedded bundled sources when set.
17    pub bundled_directory: Option<PathBuf>,
18    pub installed_directory: PathBuf,
19    pub additional_files: Vec<PathBuf>,
20}
21
22impl IndicatorCatalogPaths {
23    fn from_data_directory() -> Self {
24        Self {
25            bundled_directory: None,
26            installed_directory: default_installed_directory(),
27            additional_files: Vec::new(),
28        }
29    }
30}
31
32fn default_installed_directory() -> PathBuf {
33    std::env::var_os("FLOWSURFACE_INDICATORS_PATH")
34        .map(PathBuf::from)
35        .unwrap_or_else(|| data_path(Some("indicators")))
36}
37
38fn installed_directory() -> Result<PathBuf, String> {
39    INSTALLED_DIRECTORY
40        .get()
41        .cloned()
42        .ok_or_else(|| "Indicator catalog is not initialized".to_owned())
43}
44
45fn installed_sources(directory: &Path) -> Result<Vec<language::LoadedIndicatorSource>, String> {
46    let (files, _) = language::indicator_files_in_directory(directory)?;
47    Ok(language::load_from_files_report(&files, language::PackageOrigin::Installed).loaded_sources)
48}
49
50pub fn write_indicator_source(id: &super::IndicatorId, source: &[u8]) -> Result<(), String> {
51    let directory = installed_directory()?;
52    if super::catalog()
53        .get(id)
54        .is_some_and(|registration| registration.origin == language::PackageOrigin::Bundled)
55    {
56        return Err("Bundled indicators cannot be replaced".to_owned());
57    }
58
59    fs::create_dir_all(&directory).map_err(|error| error.to_string())?;
60    if installed_sources(&directory)?
61        .iter()
62        .any(|loaded| loaded.id == id.as_str())
63    {
64        return Err(
65            "An indicator with this ID already exists in the indicators directory".to_owned(),
66        );
67    }
68
69    let (namespace, name) = id
70        .as_str()
71        .rsplit_once('.')
72        .expect("Validated indicator ID has a namespace");
73    let destination_directory = directory.join(namespace);
74    let destination = destination_directory.join(format!("{name}.ind"));
75
76    let source_text = std::str::from_utf8(source).map_err(|error| error.to_string())?;
77    let report = language::load_compiled_report(
78        [(
79            destination.clone(),
80            indicator_engine::compile(source_text, &destination),
81        )],
82        language::PackageOrigin::Installed,
83    );
84    if !report.diagnostics.is_empty() {
85        return Err(report.diagnostics.join("\n"));
86    }
87    if report.registrations.len() != 1 || report.registrations[0].id != *id {
88        return Err("Downloaded indicator does not declare the expected ID".to_owned());
89    }
90
91    fs::create_dir_all(&destination_directory).map_err(|error| error.to_string())?;
92    // A non-source extension keeps interrupted writes out of source discovery on the same filesystem.
93    let temporary = destination_directory.join(format!(".install-{}.part", uuid::Uuid::new_v4()));
94    let mut file = fs::OpenOptions::new()
95        .write(true)
96        .create_new(true)
97        .open(&temporary)
98        .map_err(|error| error.to_string())?;
99    let result = (|| {
100        use std::io::Write;
101        file.write_all(source).map_err(|error| error.to_string())?;
102        file.sync_all().map_err(|error| error.to_string())?;
103        drop(file);
104
105        // A hard link publishes the complete file atomically and refuses to replace an existing path.
106        fs::hard_link(&temporary, &destination).map_err(|error| error.to_string())
107    })();
108    if let Err(error) = fs::remove_file(&temporary) {
109        log::warn!("Could not remove indicator installation staging file: {error}");
110    }
111    result
112}
113
114pub fn remove_indicator_source(id: &super::IndicatorId) -> Result<(), String> {
115    let directory = installed_directory()?;
116    if super::catalog()
117        .get(id)
118        .is_some_and(|registration| registration.origin == language::PackageOrigin::Bundled)
119    {
120        return Err("Bundled indicators cannot be removed".to_owned());
121    }
122
123    let installed = installed_sources(&directory)?;
124    let mut sources = installed.iter().filter(|source| source.id == id.as_str());
125    let source = sources.next().ok_or_else(|| {
126        "No matching source was found in the configured indicators directory".to_owned()
127    })?;
128    if sources.next().is_some() {
129        return Err(
130            "Multiple sources declare this indicator ID; remove the conflicting files manually"
131                .to_owned(),
132        );
133    }
134    fs::remove_file(&source.path).map_err(|error| error.to_string())
135}
136
137pub fn initialize_indicator_catalog() -> Result<(), String> {
138    let additional_files =
139        indicator_paths_from_args().map_err(|error| format!("Invalid indicator path: {error}"))?;
140    let mut paths = IndicatorCatalogPaths::from_data_directory();
141    paths.additional_files.extend(additional_files);
142
143    initialize_indicator_catalog_with_paths(&paths)
144        .map_err(|error| format!("Failed to load indicators: {error}"))
145}
146
147pub fn initialize_indicator_catalog_with_paths(
148    paths: &IndicatorCatalogPaths,
149) -> Result<(), String> {
150    let _ = INSTALLED_DIRECTORY.set(paths.installed_directory.clone());
151    let sources = load_catalog_sources(paths)?;
152    let bundled_ids = sources
153        .bundled
154        .iter()
155        .map(|registration| registration.id.clone())
156        .collect::<Vec<_>>();
157
158    super::install_catalog_with_bundled_diagnostics(
159        sources.bundled,
160        sources.installed,
161        sources.diagnostics,
162    )
163    .map_err(|error| error.to_string())?;
164    super::bundled::validate_bundled_catalog(&bundled_ids)
165        .map_err(|error| format!("Failed to load bundled indicators: {error}"))
166}
167
168fn indicator_paths_from_args() -> Result<Vec<PathBuf>, String> {
169    let args = std::env::args().skip(1).collect::<Vec<_>>();
170    let mut paths = Vec::new();
171    let mut index = 0;
172    while index < args.len() {
173        if args[index] == "--indicator-path" {
174            let path = args
175                .get(index + 1)
176                .ok_or_else(|| "--indicator-path requires an explicit .ind file".to_owned())?;
177            paths.push(validate_indicator_path(PathBuf::from(path))?);
178            index += 2;
179            continue;
180        } else if let Some(path) = args[index].strip_prefix("--indicator-path=") {
181            if path.trim().is_empty() {
182                return Err("--indicator-path requires an explicit .ind file".to_owned());
183            }
184            paths.push(validate_indicator_path(PathBuf::from(path))?);
185        }
186        index += 1;
187    }
188    Ok(paths)
189}
190
191fn validate_indicator_path(path: PathBuf) -> Result<PathBuf, String> {
192    if path.is_dir() {
193        return Err(format!(
194            "{} is a directory; --indicator-path accepts an explicit .ind file",
195            path.display()
196        ));
197    }
198    if path.extension().and_then(|extension| extension.to_str()) != Some("ind") {
199        return Err(format!(
200            "{}: --indicator-path accepts only .ind source files",
201            path.display()
202        ));
203    }
204    Ok(path)
205}
206
207#[derive(Debug)]
208struct IndicatorCatalogSources {
209    bundled: Vec<IndicatorRegistration>,
210    installed: Vec<IndicatorRegistration>,
211    diagnostics: Vec<String>,
212}
213
214fn load_catalog_sources(paths: &IndicatorCatalogPaths) -> Result<IndicatorCatalogSources, String> {
215    let bundled_report = match &paths.bundled_directory {
216        Some(directory) => {
217            let (files, diagnostics) = language::indicator_files_in_directory(directory)?;
218            let mut report =
219                language::load_from_files_report(&files, language::PackageOrigin::Bundled);
220            report.diagnostics.extend(diagnostics);
221            report
222        }
223        None => super::bundled::load_bundled_sources(),
224    };
225    if !bundled_report.diagnostics.is_empty() {
226        return Err(format!(
227            "bundled indicator source validation failed:\n{}",
228            bundled_report.diagnostics.join("\n")
229        ));
230    }
231    if bundled_report.registrations.is_empty() {
232        return Err("bundled indicator collection contains no valid sources".to_owned());
233    }
234
235    let mut diagnostics = Vec::new();
236    let installed_files = if let Err(error) = fs::create_dir_all(&paths.installed_directory) {
237        diagnostics.push(format!(
238            "could not create installed indicator directory {}: {error}",
239            paths.installed_directory.display()
240        ));
241        Vec::new()
242    } else {
243        match language::indicator_files_in_directory(&paths.installed_directory) {
244            Ok((files, discovery_diagnostics)) => {
245                diagnostics.extend(discovery_diagnostics);
246                files
247            }
248            Err(error) => {
249                diagnostics.push(error);
250                Vec::new()
251            }
252        }
253    };
254    let mut installed_report =
255        language::load_from_files_report(&installed_files, language::PackageOrigin::Installed);
256    let additional_report =
257        language::load_from_files_report(&paths.additional_files, language::PackageOrigin::Bundled);
258    installed_report
259        .registrations
260        .extend(additional_report.registrations);
261    installed_report
262        .diagnostics
263        .extend(additional_report.diagnostics);
264
265    for diagnostic in &installed_report.diagnostics {
266        warn!("Ignoring invalid installed indicator source: {diagnostic}");
267    }
268    diagnostics.extend(installed_report.diagnostics);
269
270    Ok(IndicatorCatalogSources {
271        bundled: bundled_report.registrations,
272        installed: installed_report.registrations,
273        diagnostics,
274    })
275}
276
277#[cfg(test)]
278mod tests {
279    use super::super::{catalog, language::PackageOrigin};
280    use super::*;
281
282    fn source(id: &str) -> String {
283        format!(
284            "language(1)\nindicator(\"Example\", id = {id:?})\nformat(precision = \"base_ticker_min_tick\")\nplot(chart.close)"
285        )
286    }
287
288    #[test]
289    fn origins_and_full_ids_control_collection_loading() {
290        let root =
291            std::env::temp_dir().join(format!("flowsurface-host-loading-{}", std::process::id()));
292        let bundled = root.join("bundled");
293        let installed = root.join("installed");
294        std::fs::create_dir_all(&bundled).unwrap();
295        std::fs::create_dir_all(installed.join("alice")).unwrap();
296        std::fs::create_dir_all(installed.join("bob")).unwrap();
297        std::fs::write(bundled.join("rsi.ind"), source("catalog.rsi")).unwrap();
298        std::fs::write(installed.join("alice/rsi.ind"), source("core.rsi")).unwrap();
299        std::fs::write(installed.join("bob/rsi.ind"), source("bob.rsi")).unwrap();
300        std::fs::write(installed.join("broken.ind"), "language(1)\nindicator(").unwrap();
301        let paths = IndicatorCatalogPaths {
302            bundled_directory: Some(bundled.clone()),
303            installed_directory: installed,
304            additional_files: Vec::new(),
305        };
306        let sources = load_catalog_sources(&paths).unwrap();
307        assert_eq!(sources.bundled.len(), 1);
308        assert_eq!(sources.installed.len(), 2);
309        assert_eq!(sources.bundled[0].origin, PackageOrigin::Bundled);
310        assert!(
311            sources
312                .installed
313                .iter()
314                .all(|registration| registration.origin == PackageOrigin::Installed)
315        );
316        assert!(!sources.diagnostics.is_empty());
317        std::fs::write(bundled.join("broken.ind"), "language(1)\nindicator(").unwrap();
318        assert!(
319            load_catalog_sources(&paths)
320                .unwrap_err()
321                .contains("bundled indicator source validation failed")
322        );
323        std::fs::remove_dir_all(root).unwrap();
324    }
325
326    #[test]
327    fn embedded_bundle_initializes_without_a_source_directory() {
328        let root =
329            std::env::temp_dir().join(format!("flowsurface-host-embedded-{}", std::process::id()));
330        let paths = IndicatorCatalogPaths {
331            bundled_directory: None,
332            installed_directory: root.clone(),
333            additional_files: Vec::new(),
334        };
335        initialize_indicator_catalog_with_paths(&paths).unwrap();
336        let report = super::super::bundled::load_bundled_sources();
337        assert!(!report.registrations.is_empty());
338        assert_eq!(catalog().registrations().len(), report.registrations.len());
339        assert!(
340            catalog()
341                .registrations()
342                .all(|registration| registration.origin == PackageOrigin::Bundled)
343        );
344        std::fs::remove_dir_all(root).unwrap();
345    }
346}
flowsurface_data/chart/kline/indicators/loader.rs

flowsurface_data/chart/kline/indicators/
loader.rs
