flowsurface_data/chart/kline/indicators/
manifest.rs

1use super::{IndicatorId, catalog, language::PackageOrigin};
2use serde::Deserialize;
3use std::{
4    collections::BTreeSet,
5    time::{Duration, Instant},
6};
7
8/// Indicator sources published by the catalog service, as the host may install them.
9#[derive(Debug, Clone, Deserialize)]
10pub struct Catalog {
11    schema_version: u32,
12    language_version: u32,
13    source_revision: String,
14    source_base_path: String,
15    pub indicators: Vec<Entry>,
16}
17
18#[derive(Debug, Clone, Deserialize)]
19pub struct Entry {
20    pub id: IndicatorId,
21    pub name: String,
22    pub description: Option<String>,
23    pub author: Option<Author>,
24    pub bases: Vec<String>,
25    pub path: String,
26    pub sha256: String,
27    pub size_bytes: usize,
28}
29
30#[derive(Debug, Clone, Deserialize)]
31pub struct Author {
32    pub name: String,
33}
34
35#[derive(Debug, Clone)]
36pub struct FetchError {
37    pub message: String,
38    retry_at: Option<Instant>,
39}
40
41impl FetchError {
42    pub fn retry_after(message: impl Into<String>, delay: Duration) -> Self {
43        Self {
44            message: message.into(),
45            retry_at: Some(Instant::now() + delay),
46        }
47    }
48}
49
50impl From<String> for FetchError {
51    fn from(message: String) -> Self {
52        Self {
53            message,
54            retry_at: None,
55        }
56    }
57}
58
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60pub enum SourceChange {
61    Installed,
62    Removed,
63}
64
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum Installation {
67    Bundled,
68    Installed,
69    NotInstalled,
70}
71
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub enum Trigger {
74    /// Loads only while the catalog has never been requested.
75    Open,
76    /// Reloads on request, subject to the retry cooldown.
77    Refresh,
78}
79
80/// Effect the host performs once a toggle is accepted.
81#[derive(Debug, Clone)]
82pub enum SourceAction {
83    Install { entry: Entry, base: String },
84    Remove(IndicatorId),
85}
86
87#[derive(Debug, Clone)]
88pub enum Message {
89    Open,
90    Refresh,
91    ToggleSource(IndicatorId),
92    Fetched(Result<Catalog, FetchError>),
93    SourceChanged {
94        id: IndicatorId,
95        change: SourceChange,
96        result: Result<(), FetchError>,
97    },
98}
99
100#[derive(Default)]
101pub struct State {
102    catalog: Option<Catalog>,
103    status: Status,
104    pending: BTreeSet<IndicatorId>,
105    operation: SourceOperation,
106}
107
108#[derive(Default)]
109enum Status {
110    #[default]
111    Unopened,
112    Loading,
113    Ready,
114    Failed(FetchError),
115}
116
117#[derive(Default)]
118enum SourceOperation {
119    #[default]
120    Idle,
121    Running,
122    Failed {
123        id: IndicatorId,
124        error: FetchError,
125    },
126}
127
128impl Catalog {
129    pub fn parse(bytes: &[u8]) -> Result<Self, String> {
130        let mut catalog: Self =
131            serde_json::from_slice(bytes).map_err(|error| format!("Invalid catalog: {error}"))?;
132        catalog.validate()?;
133        catalog
134            .indicators
135            .sort_by_cached_key(|entry| entry.name.to_lowercase());
136        Ok(catalog)
137    }
138
139    pub fn source_base_path(&self) -> &str {
140        &self.source_base_path
141    }
142
143    fn validate(&self) -> Result<(), String> {
144        if self.schema_version != 1 || self.language_version != 1 {
145            return Err(
146                "This catalog requires an unsupported format or language version".to_owned(),
147            );
148        }
149
150        if self.source_revision.len() != 40
151            || !self
152                .source_revision
153                .bytes()
154                .all(|byte| byte.is_ascii_hexdigit())
155            || self.source_base_path != format!("/v1/sources/{}/", self.source_revision)
156        {
157            return Err("Invalid catalog source revision".to_owned());
158        }
159
160        let mut ids = BTreeSet::new();
161        for entry in &self.indicators {
162            if !ids.insert(&entry.id) {
163                return Err("Catalog contains duplicate IDs".to_owned());
164            }
165
166            if !entry.path.ends_with(".ind")
167                || entry.path.split('/').any(|part| {
168                    part.is_empty()
169                        || part == "."
170                        || part == ".."
171                        || part.contains(['\\', ':', '?', '#', '%'])
172                })
173            {
174                return Err(format!("Invalid source path for {}", entry.id.as_str()));
175            }
176
177            if entry.size_bytes == 0
178                || entry.size_bytes > indicator_engine::CompileLimits::default().max_source_bytes
179                || entry.sha256.len() != 64
180                || !entry.sha256.bytes().all(|byte| byte.is_ascii_hexdigit())
181            {
182                return Err(format!(
183                    "Invalid source size or checksum for {}",
184                    entry.id.as_str()
185                ));
186            }
187        }
188        Ok(())
189    }
190}
191
192impl State {
193    pub fn catalog(&self) -> Option<&Catalog> {
194        self.catalog.as_ref()
195    }
196
197    pub fn is_loading(&self) -> bool {
198        matches!(self.status, Status::Loading)
199    }
200
201    pub fn error(&self) -> Option<&FetchError> {
202        match &self.status {
203            Status::Failed(error) => Some(error),
204            _ => None,
205        }
206    }
207
208    pub fn error_for(&self, id: &IndicatorId) -> Option<&FetchError> {
209        match &self.operation {
210            SourceOperation::Failed { id: failed, error } if failed == id => Some(error),
211            _ => None,
212        }
213    }
214
215    pub fn installation(&self, entry: &Entry) -> Installation {
216        let origin = catalog()
217            .get(&entry.id)
218            .map(|registration| registration.origin);
219        if origin == Some(PackageOrigin::Bundled) {
220            return Installation::Bundled;
221        }
222
223        let pending = self.pending.contains(&entry.id);
224        let present = match origin {
225            Some(_) => !pending,
226            None => pending,
227        };
228        if present {
229            Installation::Installed
230        } else {
231            Installation::NotInstalled
232        }
233    }
234
235    pub fn restart_pending(&self) -> bool {
236        !self.pending.is_empty()
237    }
238
239    pub fn restart_pending_for(&self, id: &IndicatorId) -> bool {
240        self.pending.contains(id)
241    }
242
243    pub fn can_refresh(&self) -> bool {
244        !self.is_loading() && self.can_request()
245    }
246
247    pub fn can_toggle(&self, installation: Installation) -> bool {
248        !matches!(installation, Installation::Bundled) && !self.is_running() && self.can_request()
249    }
250
251    pub fn begin_load(&mut self, trigger: Trigger) -> bool {
252        let allowed = match trigger {
253            Trigger::Open => self.is_unopened() && self.can_request(),
254            Trigger::Refresh => self.can_refresh(),
255        };
256        if allowed {
257            self.status = Status::Loading;
258        }
259        allowed
260    }
261
262    pub fn loaded(&mut self, catalog: Catalog) {
263        self.catalog = Some(catalog);
264        self.status = Status::Ready;
265    }
266
267    pub fn failed(&mut self, error: FetchError) {
268        self.status = Status::Failed(error);
269    }
270
271    pub fn begin_toggle(&mut self, id: &IndicatorId) -> Option<SourceAction> {
272        let manifest = self.catalog.as_ref()?;
273        let entry = manifest
274            .indicators
275            .iter()
276            .find(|entry| entry.id == *id)?
277            .clone();
278        let base = manifest.source_base_path().to_owned();
279
280        let installation = self.installation(&entry);
281        if !self.can_toggle(installation) {
282            return None;
283        }
284        self.operation = SourceOperation::Running;
285
286        Some(if matches!(installation, Installation::Installed) {
287            SourceAction::Remove(entry.id)
288        } else {
289            SourceAction::Install { entry, base }
290        })
291    }
292
293    pub fn operation_succeeded(&mut self, id: IndicatorId, change: SourceChange) {
294        self.record_change(id, change);
295        self.operation = SourceOperation::Idle;
296    }
297
298    pub fn operation_failed(&mut self, id: IndicatorId, error: FetchError) {
299        self.operation = SourceOperation::Failed { id, error };
300    }
301
302    fn is_unopened(&self) -> bool {
303        matches!(self.status, Status::Unopened)
304    }
305
306    fn is_running(&self) -> bool {
307        matches!(self.operation, SourceOperation::Running)
308    }
309
310    fn can_request(&self) -> bool {
311        let catalog_retry = self.error().and_then(|error| error.retry_at);
312        let operation_retry = match &self.operation {
313            SourceOperation::Failed { error, .. } => error.retry_at,
314            _ => None,
315        };
316        catalog_retry
317            .into_iter()
318            .chain(operation_retry)
319            .max()
320            .is_none_or(|at| Instant::now() >= at)
321    }
322
323    fn record_change(&mut self, id: IndicatorId, change: SourceChange) {
324        let loaded = catalog().get(&id).is_some();
325        let pending = match change {
326            SourceChange::Installed => !loaded,
327            SourceChange::Removed => loaded,
328        };
329        if pending {
330            self.pending.insert(id);
331        } else {
332            self.pending.remove(&id);
333        }
334    }
335}
flowsurface_data/chart/kline/indicators/manifest.rs

flowsurface_data/chart/kline/indicators/
manifest.rs
