phemios_core/
network_snapshot.rs

1//! Complete same-epoch presentation, bounded before retaining another page.
2use super::Client;
3use serde_json::{Value, json};
4use std::io::{self, Write};
5
6// Count serialized bytes without constructing a second page-sized JSON buffer.
7struct SnapshotBudget {
8    remaining: u64,
9}
10impl Write for SnapshotBudget {
11    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
12        self.remaining = self
13            .remaining
14            .checked_sub(bytes.len() as u64)
15            .ok_or_else(|| io::Error::other("catalogue limit exceeded"))?;
16        Ok(bytes.len())
17    }
18    fn flush(&mut self) -> io::Result<()> {
19        Ok(())
20    }
21}
22impl SnapshotBudget {
23    fn reserve(&mut self, value: &Value) -> Result<(), String> {
24        serde_json::to_writer(self, value).map_err(|_| "Catalogue trop grand pour ce cache.".into())
25    }
26}
27/// Admission holds no network lock and also discards a reply retired during its request.
28fn admitted<T>(
29    current: &impl Fn() -> Result<(), String>,
30    work: impl FnOnce() -> Result<T, String>,
31) -> Result<T, String> {
32    current()?;
33    let result = work();
34    current()?;
35    result
36}
37impl Client {
38    /// Synchronize complete same-epoch public projections under the cache's shared JSON-byte limit.
39    /// Oversized pages fail before retention; differing projection epochs allow three attempts.
40    pub fn snapshot(&self) -> Result<Value, String> {
41        self.snapshot_admitted(|| Ok(()))
42    }
43    /// Check the caller's exact session before/after each policy, page and detail request.
44    /// Retiring admission prevents follow-up IO; an already dispatched request may finish.
45    pub fn snapshot_admitted(
46        &self,
47        current: impl Fn() -> Result<(), String>,
48    ) -> Result<Value, String> {
49        for _ in 0..3 {
50            let policy = admitted(&current, || self.parental_status_admitted(&current))?;
51            let mut budget = SnapshotBudget {
52                remaining: crate::cache::MAX_CATALOG,
53            };
54            // Reserve the fixed wrapper first; page-array overhead is deliberately conservative.
55            budget.reserve(&json!({"artists":[],"albums":[],"tracks":[],"labels":[],"scannedAt":u64::MAX,"server":self.server,"accountId":self.account["id"]}))?;
56            if let Some(status) = policy
57                .as_ref()
58                .filter(|status| status["policy"]["enabled"] == true)
59            {
60                budget.reserve(&json!({"parentalEnabled": true, "parentalRevision":status["revision"], "parentalPolicy":status["policy"]}))?;
61            }
62            let artists = self.pages("/api/v1/presentation/artists", &mut budget, &current)?;
63            let albums = self.pages("/api/v1/presentation/albums", &mut budget, &current)?;
64            let mut tracks = self.pages("/api/v1/presentation/tracks", &mut budget, &current)?;
65            for track in &mut tracks.0 {
66                let fields = track.as_object_mut().ok_or("Piste Aède invalide.")?;
67                fields.retain(|key, _| {
68                    matches!(
69                        key.as_str(),
70                        "reference"
71                            | "title"
72                            | "release"
73                            | "album_title"
74                            | "disc_no"
75                            | "track_no"
76                            | "album_position"
77                            | "duration_ms"
78                            | "analysis_available"
79                            | "lyrics_available"
80                            | "spectrum_available"
81                    )
82                });
83            }
84            let summaries = self.pages("/api/v1/labels", &mut budget, &current)?;
85            if artists.1 == albums.1 && artists.1 == tracks.1 && artists.1 == summaries.1 {
86                let labels = self.label_details(summaries.0, &mut budget, &current)?;
87                if !labels.is_empty() {
88                    // Detail responses have no epoch: a final label probe refuses torn associations.
89                    let probe = admitted(&current, || {
90                        self.request_admitted(
91                            "GET",
92                            "/api/v1/labels",
93                            &[("offset", "0"), ("limit", "1")],
94                            None,
95                            &current,
96                        )
97                    })?;
98                    if probe["scanned_at"].as_u64() != Some(artists.1)
99                        || probe["total"].as_u64() != Some(labels.len() as u64)
100                    {
101                        continue;
102                    }
103                }
104                if admitted(&current, || self.parental_status_admitted(&current))?
105                    .as_ref()
106                    .map(|v| (&v["revision"], &v["policy"]))
107                    != policy.as_ref().map(|v| (&v["revision"], &v["policy"]))
108                {
109                    continue;
110                }
111                let mut snapshot = json!({"artists":artists.0,"albums":albums.0,"tracks":tracks.0,"labels":labels,"scannedAt":artists.1,"server":self.server,"accountId":self.account["id"]});
112                if let Some(policy) = policy.filter(|policy| policy["policy"]["enabled"] == true) {
113                    snapshot["parentalEnabled"] = Value::Bool(true);
114                    snapshot["parentalRevision"] = policy["revision"].clone();
115                    snapshot["parentalPolicy"] = policy["policy"].clone();
116                }
117                return Ok(snapshot);
118            }
119        }
120        Err("Le catalogue change pendant la synchronisation. Réessayez.".into())
121    }
122    fn label_details(
123        &self,
124        summaries: Vec<Value>,
125        budget: &mut SnapshotBudget,
126        current: &impl Fn() -> Result<(), String>,
127    ) -> Result<Vec<Value>, String> {
128        let mut labels = Vec::with_capacity(summaries.len());
129        for summary in summaries {
130            let reference = summary["reference"]
131                .as_str()
132                .ok_or("Label Aède invalide.")?;
133            let detail = admitted(current, || {
134                self.request_admitted("GET", "/api/v1/label", &[("ref", reference)], None, current)
135            })?;
136            if detail["reference"] != summary["reference"]
137                || detail["name"] != summary["name"]
138                || !detail["releases"].is_array()
139                || summary["release_count"].as_u64()
140                    != detail["releases"].as_array().map(|rows| rows.len() as u64)
141            {
142                return Err("Label Aède invalide.".into());
143            }
144            // Copy only this public contract, never arbitrary response extensions or owner data.
145            let mut label = json!({"reference":detail["reference"], "name":detail["name"],
146                "release_count":summary["release_count"], "releases":detail["releases"]});
147            if !detail["biography"].is_null() {
148                let prose = &detail["biography"];
149                if ["text", "url", "lang", "licence", "source"]
150                    .iter()
151                    .any(|key| prose[key].as_str().is_none())
152                {
153                    return Err("Biographie Aède invalide.".into());
154                }
155                label["biography"] = json!({"text":prose["text"], "url":prose["url"], "lang":prose["lang"], "licence":prose["licence"], "source":prose["source"]});
156            }
157            budget.reserve(&label)?;
158            labels.push(label);
159        }
160        Ok(labels)
161    }
162    fn pages(
163        &self,
164        path: &str,
165        budget: &mut SnapshotBudget,
166        current: &impl Fn() -> Result<(), String>,
167    ) -> Result<(Vec<Value>, u64), String> {
168        let mut items = Vec::new();
169        let mut stamp = None;
170        let mut total = None;
171        loop {
172            let offset = items.len().to_string();
173            let mut page = admitted(current, || {
174                self.request_admitted(
175                    "GET",
176                    path,
177                    &[("offset", &offset), ("limit", "200")],
178                    None,
179                    current,
180                )
181            })?;
182            let count = page["total"]
183                .as_u64()
184                .and_then(|count| usize::try_from(count).ok())
185                .filter(|count| *count <= 1_000_000)
186                .ok_or("Pagination Aède invalide.")?;
187            let at = page["scanned_at"]
188                .as_u64()
189                .ok_or("Pagination Aède invalide.")?;
190            if stamp.is_some_and(|previous| previous != at)
191                || total.is_some_and(|previous| previous != count)
192            {
193                return Err("Le catalogue a changé. Relancez la synchronisation.".into());
194            }
195            stamp = Some(at);
196            total = Some(count);
197            let rows = page["items"]
198                .as_array()
199                .ok_or("Pagination Aède invalide.")?;
200            if rows.len() > 200
201                || page["offset"].as_u64() != Some(items.len() as u64)
202                || items.len() + rows.len() > count
203                || rows.is_empty() && items.len() < count
204            {
205                return Err("Pagination Aède incohérente.".into());
206            }
207            budget.reserve(&page["items"])?;
208            items.append(
209                page.get_mut("items")
210                    .and_then(Value::as_array_mut)
211                    .ok_or("Pagination Aède invalide.")?,
212            );
213            if items.len() == count {
214                return Ok((items, at));
215            }
216        }
217    }
218}
phemios_core/network_snapshot.rs

phemios_core/
network_snapshot.rs
