aven_core/sync/
wire.rs

1use std::collections::HashSet;
2
3use anyhow::{Context, Result, bail};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6
7use crate::attachments::validation::{
8    MAX_BLOB_BYTES, validate_alt_text, validate_blob_size, validate_dimensions, validate_filename,
9    validate_media_type,
10};
11use crate::change_log::op_type;
12use crate::ids::{BASE32, MetadataFieldId, ProjectId, TaskId, WorkspaceId};
13use crate::task_fields::TaskField;
14
15mod changes;
16mod recurrence;
17#[cfg(test)]
18mod test_support;
19#[cfg(test)]
20mod tests;
21
22pub const SYNC_PROTOCOL_VERSION: u32 = 19;
23const MAX_CHANGE_PAYLOAD_BYTES: usize = 64 * 1024;
24
25pub(crate) fn serialize_change_payload(payload: &Value) -> Result<String> {
26    let serialized = serde_json::to_string(payload)?;
27    if serialized.len() > MAX_CHANGE_PAYLOAD_BYTES {
28        return Err(crate::error::CoreError::validation(format!(
29            "error invalid-sync-change payload-too-large limit={MAX_CHANGE_PAYLOAD_BYTES}"
30        ))
31        .into());
32    }
33    Ok(serialized)
34}
35
36pub fn sync_server_url_is_valid(server: &str) -> bool {
37    url::Url::parse(server)
38        .as_ref()
39        .is_ok_and(sync_server_url_is_valid_url)
40}
41
42fn sync_server_url_is_valid_url(url: &url::Url) -> bool {
43    matches!(url.scheme(), "http" | "https")
44        && url.host_str().is_some()
45        && url.username().is_empty()
46        && url.password().is_none()
47        && url.query().is_none()
48        && url.fragment().is_none()
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize)]
52pub struct ChangeWire {
53    pub change_id: String,
54    pub client_id: String,
55    pub local_seq: i64,
56    pub entity_type: String,
57    pub entity_id: String,
58    pub field: Option<String>,
59    pub op_type: String,
60    pub payload: Value,
61    pub base_version: Option<String>,
62    pub created_at: String,
63    pub server_seq: Option<i64>,
64}
65
66pub const MAX_MOVE_TASKS: usize = 256;
67
68#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
69#[serde(deny_unknown_fields)]
70pub(crate) struct MoveTaskMetadataSnapshot {
71    pub(crate) field_id: MetadataFieldId,
72    pub(crate) key: String,
73    pub(crate) value: Option<String>,
74    pub(crate) version: Option<String>,
75}
76
77#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
78#[serde(deny_unknown_fields)]
79pub(crate) struct MoveTaskSnapshot {
80    pub(crate) task_id: TaskId,
81    pub(crate) labels: Vec<String>,
82    pub(crate) metadata: Vec<MoveTaskMetadataSnapshot>,
83}
84
85#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
86#[serde(deny_unknown_fields)]
87pub(crate) struct MoveDependencySnapshot {
88    pub(crate) task_id: TaskId,
89    pub(crate) depends_on_task_id: TaskId,
90    pub(crate) created_at: String,
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
94#[serde(deny_unknown_fields)]
95pub(crate) struct MoveRelatedSnapshot {
96    pub(crate) task_a_id: TaskId,
97    pub(crate) task_b_id: TaskId,
98    pub(crate) linked: bool,
99    pub(crate) last_change_id: String,
100}
101
102#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
103#[serde(deny_unknown_fields)]
104pub(crate) struct MoveEpicSnapshot {
105    pub(crate) child_task_id: TaskId,
106    pub(crate) epic_task_id: TaskId,
107    pub(crate) created_at: String,
108}
109
110#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
111#[serde(deny_unknown_fields)]
112pub(crate) struct MoveTasksPayload {
113    pub(crate) source_workspace_id: WorkspaceId,
114    pub(crate) target_workspace_id: WorkspaceId,
115    pub(crate) project_id: ProjectId,
116    pub(crate) project_key: String,
117    pub(crate) project_name: String,
118    pub(crate) project_prefix: String,
119    pub(crate) tasks: Vec<MoveTaskSnapshot>,
120    pub(crate) dependencies: Vec<MoveDependencySnapshot>,
121    pub(crate) related: Vec<MoveRelatedSnapshot>,
122    pub(crate) epics: Vec<MoveEpicSnapshot>,
123}
124
125impl MoveTasksPayload {
126    pub(crate) fn from_change(change: &ChangeWire) -> Result<Self> {
127        changes::validate_move_tasks(change)
128    }
129}
130
131#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
132pub(crate) struct AttachmentAddPayload {
133    pub(crate) workspace_id: String,
134    pub(crate) workspace_key: String,
135    pub(crate) attachment_id: String,
136    pub(crate) sha256: String,
137    pub(crate) byte_size: i64,
138    pub(crate) media_type: String,
139    pub(crate) filename: Option<String>,
140    pub(crate) alt_text: Option<String>,
141    pub(crate) width: Option<i64>,
142    pub(crate) height: Option<i64>,
143    pub(crate) created_at: String,
144}
145
146impl AttachmentAddPayload {
147    pub(crate) fn from_change(change: &ChangeWire) -> Result<Self> {
148        validate_attachment_change_envelope(change, op_type::ATTACHMENT_ADD)?;
149        required_workspace_payload(&change.payload)?;
150        let attachment_id = required_string_payload("attachment_id", &change.payload)?;
151        ensure_sync_id("attachment_id", &attachment_id)?;
152        let sha256 = required_string_payload("sha256", &change.payload)?;
153        validate_sha256_for_sync(&sha256)?;
154        let byte_size = required_i64_payload("byte_size", &change.payload)?;
155        validate_blob_size_for_sync(byte_size)?;
156        let media_type = required_string_payload("media_type", &change.payload)?;
157        map_attachment_validation(validate_media_type(&media_type))?;
158        let filename = optional_string_payload("filename", &change.payload)?;
159        map_attachment_validation(validate_filename(filename.as_deref()))?;
160        let alt_text = optional_string_payload("alt_text", &change.payload)?;
161        map_attachment_validation(validate_alt_text(alt_text.as_deref()))?;
162        let width = optional_i64_payload("width", &change.payload)?;
163        let height = optional_i64_payload("height", &change.payload)?;
164        map_attachment_validation(validate_dimensions(width, height))?;
165        required_timestamp_payload("created_at", &change.payload)?;
166        deserialize_attachment_payload(&change.payload)
167    }
168}
169
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub(crate) struct AttachmentDeletePayload {
172    pub(crate) workspace_id: String,
173    pub(crate) workspace_key: String,
174    pub(crate) attachment_id: String,
175    pub(crate) deleted_at: String,
176}
177
178impl AttachmentDeletePayload {
179    pub(crate) fn from_change(change: &ChangeWire) -> Result<Self> {
180        validate_attachment_change_envelope(change, op_type::ATTACHMENT_DELETE)?;
181        required_workspace_payload(&change.payload)?;
182        let attachment_id = required_string_payload("attachment_id", &change.payload)?;
183        ensure_sync_id("attachment_id", &attachment_id)?;
184        required_timestamp_payload("deleted_at", &change.payload)?;
185        deserialize_attachment_payload(&change.payload)
186    }
187}
188
189fn deserialize_attachment_payload<T>(payload: &Value) -> Result<T>
190where
191    T: serde::de::DeserializeOwned,
192{
193    serde_json::from_value(payload.clone())
194        .map_err(|error| anyhow::anyhow!("error invalid-sync-change payload {error}"))
195}
196
197fn validate_attachment_change_envelope(change: &ChangeWire, expected_op_type: &str) -> Result<()> {
198    if change.op_type != expected_op_type {
199        bail!("error invalid-sync-change op_type={}", change.op_type);
200    }
201    ensure_entity_type(change, "task")?;
202    ensure_sync_id("entity_id", &change.entity_id)?;
203    if change.field.as_deref() != Some("attachments") {
204        bail!("error invalid-sync-change field=attachments");
205    }
206    if !change.payload.is_object() {
207        bail!("error invalid-sync-change payload expected-object");
208    }
209    serialize_change_payload(&change.payload)?;
210    Ok(())
211}
212
213#[cfg(any(test, feature = "test-support"))]
214pub fn validate_pushed_change(change: &ChangeWire) -> Result<()> {
215    validate_local_change_shape(change)?;
216    super::protocol::validate_change(SYNC_PROTOCOL_VERSION, change)
217}
218
219/// Largest local sequence a change may carry. Installations number their own
220/// writes upward from one and adopt the largest applied remote value, so the
221/// bound keeps every replica's next local sequence far from overflow.
222pub(crate) const MAX_LOCAL_SEQ: i64 = 1 << 53;
223
224pub(crate) fn validate_local_change_shape(change: &ChangeWire) -> Result<()> {
225    ensure_non_empty("change_id", &change.change_id)?;
226    ensure_non_empty("client_id", &change.client_id)?;
227    ensure_non_empty("entity_id", &change.entity_id)?;
228    ensure_non_empty("op_type", &change.op_type)?;
229    ensure_non_empty("entity_type", &change.entity_type)?;
230    ensure_sync_id("change_id", &change.change_id)?;
231    if change.server_seq.is_some() {
232        bail!("error invalid-sync-change server_seq client-supplied");
233    }
234    if !(0..=MAX_LOCAL_SEQ).contains(&change.local_seq) {
235        bail!("error invalid-sync-change local_seq out-of-range");
236    }
237    if !change.payload.is_object() {
238        bail!("error invalid-sync-change payload expected-object");
239    }
240    serialize_change_payload(&change.payload)?;
241
242    match change.op_type.as_str() {
243        op_type::MOVE_TASKS => {
244            changes::validate_move_tasks(change)?;
245        }
246        op_type::PUBLISH_DEVICE_LABEL => changes::validate_publish_device_label(change)?,
247        op_type::CREATE_WORKSPACE => changes::validate_create_workspace(change)?,
248        op_type::SET_WORKSPACE_FIELD => changes::validate_set_workspace_field(change)?,
249        op_type::DELETE_WORKSPACE => changes::validate_delete_workspace(change)?,
250        op_type::CREATE_PROJECT => changes::validate_create_project(change)?,
251        op_type::SET_PROJECT_METADATA => changes::validate_set_project_metadata(change)?,
252        op_type::CREATE_LABEL => changes::validate_create_label(change)?,
253        op_type::CREATE_METADATA_FIELD => changes::validate_create_metadata_field(change)?,
254        op_type::SET_METADATA_FIELD => changes::validate_set_metadata_field(change)?,
255        op_type::SET_TASK_METADATA | op_type::REMOVE_TASK_METADATA => {
256            changes::validate_task_metadata(change)?
257        }
258        op_type::CREATE_TASK => changes::validate_create_task(change)?,
259        op_type::SET_FIELD | op_type::RESOLVE_FIELD => changes::validate_task_field(change)?,
260        op_type::LABEL_ADD | op_type::LABEL_REMOVE => changes::validate_label_change(change)?,
261        op_type::NOTE_ADD => changes::validate_note_add(change)?,
262        op_type::DEPENDENCY_ADD | op_type::DEPENDENCY_REMOVE => {
263            changes::validate_dependency_change(change)?
264        }
265        op_type::RELATED_ADD | op_type::RELATED_REMOVE => changes::validate_related_change(change)?,
266        op_type::EPIC_LINK_ADD | op_type::EPIC_LINK_REMOVE => {
267            changes::validate_epic_link_change(change)?
268        }
269        op_type::PROJECT_DELETE => changes::validate_project_delete(change)?,
270        op_type::SET_LABEL_NAME => changes::validate_set_label_name(change)?,
271        op_type::LABEL_DELETE => changes::validate_label_delete(change)?,
272        op_type::NOTE_EDIT => changes::validate_note_edit(change)?,
273        op_type::LABEL_RESTORE => changes::validate_label_restore(change)?,
274        op_type::NOTE_DELETE => changes::validate_note_delete(change)?,
275        op_type::SET_RECURRENCE_METADATA | op_type::REMOVE_RECURRENCE_METADATA => {
276            changes::validate_recurrence_metadata(change)?
277        }
278        op_type::CREATE_RECURRENCE_SERIES => recurrence::validate_recurrence_create(change)?,
279        op_type::UPDATE_RECURRENCE_TEMPLATE => recurrence::validate_recurrence_template(change)?,
280        op_type::PROJECT_RECURRENCE_OCCURRENCE => {
281            recurrence::validate_recurrence_projection(change)?
282        }
283        op_type::RESOLVE_RECURRENCE_OCCURRENCE => recurrence::validate_recurrence_outcome(change)?,
284        op_type::SET_RECURRENCE_STATE | op_type::STOP_RECURRENCE_SERIES => {
285            recurrence::validate_recurrence_state(change)?
286        }
287        op_type::OPEN_RECURRENCE_PAUSE => recurrence::validate_recurrence_pause(change, true)?,
288        op_type::CLOSE_RECURRENCE_PAUSE => recurrence::validate_recurrence_pause(change, false)?,
289        op_type::ATTACHMENT_ADD => {
290            AttachmentAddPayload::from_change(change)?;
291        }
292        op_type::ATTACHMENT_DELETE => {
293            AttachmentDeletePayload::from_change(change)?;
294        }
295        _ => bail!("error invalid-sync-change op_type={}", change.op_type),
296    }
297    Ok(())
298}
299
300fn validate_sha256_for_sync(value: &str) -> Result<()> {
301    if value.len() == 64
302        && value
303            .bytes()
304            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
305    {
306        Ok(())
307    } else {
308        bail!("error invalid-sync-change invalid-sha256");
309    }
310}
311
312fn validate_blob_size_for_sync(byte_size: i64) -> Result<()> {
313    let Ok(bytes) = usize::try_from(byte_size) else {
314        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
315    };
316    if bytes == 0 || bytes > MAX_BLOB_BYTES {
317        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
318    }
319    map_attachment_validation(validate_blob_size(bytes))
320}
321
322fn map_attachment_validation(result: Result<()>) -> Result<()> {
323    result.map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
324}
325
326fn validate_sync_task_field_value(field: TaskField, value: &str) -> Result<()> {
327    field
328        .validate_value(value)
329        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
330}
331
332fn ensure_entity_type(change: &ChangeWire, expected: &str) -> Result<()> {
333    if change.entity_type == expected {
334        Ok(())
335    } else {
336        bail!(
337            "error invalid-sync-change op_type={} entity_type={} expected={}",
338            change.op_type,
339            change.entity_type,
340            expected
341        )
342    }
343}
344
345fn ensure_non_empty(name: &str, value: &str) -> Result<()> {
346    if value.trim().is_empty() {
347        bail!("error invalid-sync-change {name} empty");
348    }
349    Ok(())
350}
351
352fn ensure_sync_id(name: &str, value: &str) -> Result<()> {
353    if value.len() == 16 && value.bytes().all(|byte| BASE32.contains(&byte)) {
354        Ok(())
355    } else {
356        bail!("error invalid-sync-change {name} invalid-id");
357    }
358}
359
360fn ensure_project_id(name: &str, value: &str) -> Result<()> {
361    if value.parse::<ProjectId>().is_ok() {
362        Ok(())
363    } else {
364        bail!("error invalid-sync-change {name} invalid-id");
365    }
366}
367
368fn required_string_payload(key: &str, payload: &Value) -> Result<String> {
369    payload
370        .get(key)
371        .and_then(Value::as_str)
372        .map(str::to_string)
373        .with_context(|| format!("error invalid-sync-change payload.{key} missing"))
374}
375
376fn required_i64_payload(key: &str, payload: &Value) -> Result<i64> {
377    match payload.get(key) {
378        Some(Value::Number(value)) => value
379            .as_i64()
380            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
381        Some(Value::Null) | None => bail!("error invalid-sync-change payload.{key} missing"),
382        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
383    }
384}
385
386fn required_timestamp_payload(key: &str, payload: &Value) -> Result<String> {
387    let value = required_string_payload(key, payload)?;
388    validate_timestamp_value(&format!("payload.{key}"), &value)?;
389    Ok(value)
390}
391
392fn validate_timestamp_value(label: &str, value: &str) -> Result<()> {
393    if value.ends_with('Z') && chrono::DateTime::parse_from_rfc3339(value).is_ok() {
394        Ok(())
395    } else {
396        bail!("error invalid-sync-change {label} invalid-timestamp");
397    }
398}
399
400fn required_workspace_payload(payload: &Value) -> Result<()> {
401    let id = required_string_payload("workspace_id", payload)?;
402    if id.parse::<WorkspaceId>().is_err() {
403        bail!("error invalid-sync-change workspace_id invalid-id");
404    }
405    required_string_payload("workspace_key", payload)?;
406    Ok(())
407}
408
409fn optional_workspace_payload(payload: &Value) -> Result<()> {
410    if payload.get("workspace_id").is_none() && payload.get("workspace_key").is_none() {
411        return Ok(());
412    }
413    required_workspace_payload(payload)
414}
415
416fn optional_string_payload(key: &str, payload: &Value) -> Result<Option<String>> {
417    match payload.get(key) {
418        Some(Value::String(value)) => Ok(Some(value.clone())),
419        Some(Value::Null) | None => Ok(None),
420        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
421    }
422}
423
424fn optional_i64_payload(key: &str, payload: &Value) -> Result<Option<i64>> {
425    match payload.get(key) {
426        Some(Value::Number(value)) => value
427            .as_i64()
428            .map(Some)
429            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
430        Some(Value::Null) | None => Ok(None),
431        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
432    }
433}
434
435fn optional_bool_payload(key: &str, payload: &Value) -> Result<Option<bool>> {
436    match payload.get(key) {
437        Some(Value::Bool(value)) => Ok(Some(*value)),
438        Some(Value::Null) | None => Ok(None),
439        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
440    }
441}
442
443fn string_array_payload(key: &str, payload: &Value) -> Result<Vec<String>> {
444    payload
445        .get(key)
446        .and_then(Value::as_array)
447        .with_context(|| format!("error invalid-sync-change payload.{key} invalid"))?
448        .iter()
449        .map(|value| {
450            value
451                .as_str()
452                .map(str::to_string)
453                .with_context(|| format!("error invalid-sync-change payload.{key} invalid"))
454        })
455        .collect()
456}
457
458fn optional_string_array_payload(key: &str, payload: &Value) -> Result<()> {
459    match payload.get(key) {
460        Some(Value::Array(values))
461            if values
462                .iter()
463                .all(|value| value.as_str().is_some_and(|value| !value.trim().is_empty())) =>
464        {
465            Ok(())
466        }
467        Some(Value::Null) | None => Ok(()),
468        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
469    }
470}
471
472fn ensure_metadata_field_id(name: &str, value: &str) -> Result<()> {
473    value
474        .parse::<MetadataFieldId>()
475        .map(|_| ())
476        .map_err(|_| anyhow::anyhow!("error invalid-sync-change {name} invalid-id"))
477}
478
479fn validate_metadata_key_payload(key: &str, payload: &Value) -> Result<String> {
480    let value = required_string_payload(key, payload)?;
481    let normalized = crate::metadata::normalize_metadata_key(&value)
482        .map_err(|_| anyhow::anyhow!("error invalid-sync-change payload.{key} invalid"))?;
483    if normalized != value {
484        bail!("error invalid-sync-change payload.{key} noncanonical");
485    }
486    Ok(value)
487}
488
489fn validate_metadata_value_payload(key: &str, payload: &Value) -> Result<String> {
490    let value = required_string_payload(key, payload)?;
491    if value.len() > crate::metadata::MAX_METADATA_VALUE_BYTES {
492        bail!("error invalid-sync-change payload.{key} too-large");
493    }
494    Ok(value)
495}
496
497fn validate_metadata_array_payload(payload: &Value) -> Result<()> {
498    let Some(values) = payload.get("metadata") else {
499        return Ok(());
500    };
501    let values = values
502        .as_array()
503        .context("error invalid-sync-change payload.metadata invalid")?;
504    if values.len() > crate::metadata::MAX_METADATA_VALUES {
505        bail!("error invalid-sync-change payload.metadata too-many");
506    }
507    let mut ids = HashSet::new();
508    let mut keys = HashSet::new();
509    let mut total_bytes = 0usize;
510    for value in values {
511        let object = value
512            .as_object()
513            .context("error invalid-sync-change payload.metadata invalid")?;
514        let field_id = required_string_payload("field_id", value)?;
515        ensure_metadata_field_id("metadata.field_id", &field_id)?;
516        let key = validate_metadata_key_payload("key", value)?;
517        let metadata_value = validate_metadata_value_payload("value", value)?;
518        if !ids.insert(field_id) || !keys.insert(key) {
519            bail!("error invalid-sync-change payload.metadata duplicate");
520        }
521        if object.len() != 3 {
522            bail!("error invalid-sync-change payload.metadata invalid");
523        }
524        total_bytes = total_bytes.saturating_add(metadata_value.len());
525    }
526    if total_bytes > crate::metadata::MAX_METADATA_TOTAL_BYTES {
527        bail!("error invalid-sync-change payload.metadata too-large");
528    }
529    Ok(())
530}
aven_core/sync/wire.rs

aven_core/sync/
wire.rs
