elasticctl_api/value_lists/
reads.rs

1use elasticctl_core::{Error, ErrorKind, Result, Transport, urlencode};
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4
5use super::{Container, StoredValues, ValueListType, canonical_value};
6
7#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
8pub struct BootstrapState {
9    pub list_index: bool,
10    pub list_item_index: bool,
11}
12
13impl BootstrapState {
14    pub fn ready(self) -> bool {
15        self.list_index && self.list_item_index
16    }
17}
18
19#[derive(Clone, Debug, PartialEq, Serialize)]
20pub struct RemoteContainer {
21    pub container: Container,
22    pub version: String,
23}
24
25#[derive(Clone, Debug, PartialEq, Serialize)]
26pub struct RemoteList {
27    pub container: RemoteContainer,
28    pub stored: StoredValues,
29}
30
31pub(super) fn malformed(message: impl Into<String>) -> Error {
32    Error::new(ErrorKind::Http, message)
33}
34
35pub(super) fn required_string<'a>(body: &'a Value, field: &str) -> Result<&'a str> {
36    body.get(field)
37        .and_then(Value::as_str)
38        .filter(|s| !s.is_empty())
39        .ok_or_else(|| {
40            malformed(format!(
41                "invalid value-list response field {field}: expected a non-empty string"
42            ))
43        })
44}
45
46pub(super) fn response_error(status: u16, body: &Value) -> Error {
47    Error::from_response_body(status, body.as_str().unwrap_or(&body.to_string()))
48}
49
50pub async fn bootstrap_state(t: &Transport) -> Result<BootstrapState> {
51    let (status, body) = t.get_response("/api/lists/index").await?;
52    if status == 404 {
53        return Ok(BootstrapState::default());
54    }
55    if status >= 400 {
56        return Err(response_error(status, &body));
57    }
58    for field in ["list_index", "list_item_index"] {
59        if !body.get(field).is_some_and(Value::is_boolean) {
60            return Err(malformed(format!(
61                "invalid value-list index response field {field}: expected a boolean"
62            )));
63        }
64    }
65    serde_json::from_value(body)
66        .map_err(|e| malformed(format!("invalid value-list index response: {e}")))
67}
68
69pub async fn require_bootstrapped(t: &Transport) -> Result<()> {
70    if !bootstrap_state(t).await?.ready() {
71        return Err(Error::new(
72            ErrorKind::Unsupported,
73            "value-list streams are not bootstrapped; run elasticctl value-lists init",
74        ));
75    }
76    Ok(())
77}
78
79pub fn decode_remote_container(body: &Value) -> Result<RemoteContainer> {
80    let container = Container {
81        id: required_string(body, "id")?.into(),
82        list_type: required_string(body, "type")?.to_owned().into(),
83        name: body
84            .get("name")
85            .and_then(Value::as_str)
86            .ok_or_else(|| malformed("invalid value-list name"))?
87            .into(),
88        description: body
89            .get("description")
90            .and_then(Value::as_str)
91            .ok_or_else(|| malformed("invalid value-list description"))?
92            .into(),
93        meta: match body.get("meta") {
94            Some(Value::Object(meta)) => Some(meta.clone()),
95            None => None,
96            Some(_) => return Err(malformed("invalid value-list meta: expected an object")),
97        },
98    };
99    Ok(RemoteContainer {
100        container,
101        version: required_string(body, "_version")?.into(),
102    })
103}
104
105async fn fetch_container(t: &Transport, id: &str) -> Result<Option<RemoteContainer>> {
106    let (status, body) = t
107        .get_response(&format!("/api/lists?id={}", urlencode(id)))
108        .await?;
109    if status == 404 {
110        return Ok(None);
111    }
112    if status >= 400 {
113        return Err(response_error(status, &body));
114    }
115    Ok(Some(decode_remote_container(&body)?))
116}
117
118pub async fn read_container(t: &Transport, id: &str) -> Result<Option<RemoteContainer>> {
119    let Some(remote) = fetch_container(t, id).await? else {
120        return Ok(None);
121    };
122    if remote.container.id != id {
123        return Err(malformed(format!(
124            "requested value list {id:?}, received {:?}",
125            remote.container.id
126        )));
127    }
128    Ok(Some(remote))
129}
130
131pub fn decode_export(kind: &ValueListType, text: &str) -> Result<StoredValues> {
132    kind.require_supported()?;
133    let mut values = Vec::new();
134    for (index, value) in text.split_terminator('\n').enumerate() {
135        if value.is_empty()
136            || value.chars().any(char::is_control)
137            || value.trim_matches([' ', '\t']) != value
138        {
139            return Err(malformed(format!(
140                "invalid value-list export line {}",
141                index + 1
142            )));
143        }
144        canonical_value(kind, value).map_err(|e| {
145            malformed(format!(
146                "invalid value-list export line {}: {}",
147                index + 1,
148                e.message
149            ))
150        })?;
151        values.push(value.to_owned());
152    }
153    Ok(StoredValues::new(values))
154}
155
156pub async fn export_values(t: &Transport, id: &str, kind: &ValueListType) -> Result<StoredValues> {
157    let text = t
158        .post_text(
159            &format!("/api/lists/items/_export?list_id={}", urlencode(id)),
160            None,
161        )
162        .await?;
163    decode_export(kind, &text)
164}
165
166pub async fn read_list(t: &Transport, id: &str) -> Result<Option<RemoteList>> {
167    let Some(container) = read_container(t, id).await? else {
168        return Ok(None);
169    };
170    let stored = export_values(t, id, &container.container.list_type).await?;
171    Ok(Some(RemoteList { container, stored }))
172}
173
174pub(super) async fn find_cursor(t: &Transport, base: &str, per_page: u32) -> Result<Vec<Value>> {
175    if per_page == 0 {
176        return Err(super::invalid("per_page must be positive"));
177    }
178    let mut out = Vec::new();
179    let mut page = 1u64;
180    let mut cursor: Option<String> = None;
181    let mut expected_total = None;
182    loop {
183        let separator = if base.contains('?') { '&' } else { '?' };
184        let mut path = format!("{base}{separator}page={page}&per_page={per_page}");
185        if let Some(cursor) = &cursor {
186            path.push_str(&format!("&cursor={}", urlencode(cursor)));
187        }
188        let body = t.get(&path).await?;
189        let total = body
190            .get("total")
191            .and_then(Value::as_u64)
192            .ok_or_else(|| malformed("invalid value-list _find total"))?;
193        if expected_total.is_some_and(|expected| expected != total) {
194            return Err(malformed("value-list _find total changed during read"));
195        }
196        expected_total = Some(total);
197        let data = body
198            .get("data")
199            .and_then(Value::as_array)
200            .ok_or_else(|| malformed("invalid value-list _find data"))?;
201        if body.get("page").and_then(Value::as_u64) != Some(page)
202            || body.get("per_page").and_then(Value::as_u64) != Some(u64::from(per_page))
203            || data.len() > per_page as usize
204            || out.len() as u64 + data.len() as u64 > total
205        {
206            return Err(malformed("invalid value-list _find page"));
207        }
208        let empty = data.is_empty();
209        out.extend(data.iter().cloned());
210        if out.len() as u64 == total {
211            return Ok(out);
212        }
213        if empty {
214            return Err(malformed("refusing partial value-list _find read"));
215        }
216        cursor = Some(required_string(&body, "cursor")?.into());
217        page += 1;
218    }
219}
220
221pub async fn find_containers(
222    t: &Transport,
223    filter: Option<&str>,
224    per_page: u32,
225) -> Result<Vec<RemoteContainer>> {
226    let mut path = "/api/lists/_find".to_owned();
227    if let Some(filter) = filter {
228        path.push_str(&format!("?filter={}", urlencode(filter)));
229    }
230    let mut containers: Vec<_> = find_cursor(t, &path, per_page)
231        .await?
232        .iter()
233        .map(decode_remote_container)
234        .collect::<Result<_>>()?;
235    containers.sort_by(|a, b| a.container.id.cmp(&b.container.id));
236    if containers
237        .windows(2)
238        .any(|pair| pair[0].container.id == pair[1].container.id)
239    {
240        return Err(malformed("duplicate container in value-list _find"));
241    }
242    Ok(containers)
243}
244
245pub(super) fn check_preview_container(
246    id: &str,
247    expected: Option<&RemoteContainer>,
248    observed: Option<&RemoteContainer>,
249) -> Result<()> {
250    let unchanged = match (expected, observed) {
251        (None, None) => true,
252        (Some(a), Some(b)) => {
253            a.container.id == b.container.id
254                && a.container.list_type == b.container.list_type
255                && a.version == b.version
256        }
257        _ => false,
258    };
259    if !unchanged {
260        return Err(Error::new(
261            ErrorKind::Conflict,
262            format!("value list {id} container changed since preview; re-run"),
263        ));
264    }
265    Ok(())
266}
267
268pub(super) async fn read_unchanged_list(
269    t: &Transport,
270    id: &str,
271    before: Option<&RemoteList>,
272) -> Result<Option<RemoteList>> {
273    let container = fetch_container(t, id).await?;
274    check_preview_container(id, before.map(|r| &r.container), container.as_ref())?;
275    let Some(container) = container else {
276        return Ok(None);
277    };
278    let stored = export_values(t, id, &container.container.list_type).await?;
279    Ok(Some(RemoteList { container, stored }))
280}
elasticctl_api/value_lists/reads.rs

elasticctl_api/value_lists/
reads.rs
