aven_core/sync/
protocol.rs

1//! Immutable operation contracts and replica compatibility policy.
2use anyhow::{Context, Result, bail};
3use serde_json::Value;
4use sqlx::SqliteConnection;
5
6use super::wire::{ChangeWire, SYNC_PROTOCOL_VERSION};
7use crate::db::{get_meta, set_meta};
8
9/// Ordinary releases retain this baseline, including for local-only databases.
10pub const MAINTAINED_PROTOCOL_BASELINE: u32 = 18;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub struct SyncCompatibilityError {
14    pub server_protocol: u32,
15    pub client_protocol: u32,
16}
17
18impl std::fmt::Display for SyncCompatibilityError {
19    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
20        let action = if self.server_protocol > self.client_protocol {
21            "Update Aven on this device to continue syncing."
22        } else {
23            "Update your Aven sync server to continue syncing."
24        };
25        write!(f, "{action} Your local tasks and edits remain saved.")
26    }
27}
28impl std::error::Error for SyncCompatibilityError {}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct SharedOperationCompatibilityError {
32    pub required_protocol: u32,
33    pub established_protocol: u32,
34}
35
36impl std::fmt::Display for SharedOperationCompatibilityError {
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        f.write_str("This feature requires a newer Aven sync server. Existing tasks and edits will continue syncing normally.")
39    }
40}
41impl std::error::Error for SharedOperationCompatibilityError {}
42
43pub fn supports_protocol(protocol: u32) -> bool {
44    (MAINTAINED_PROTOCOL_BASELINE..=SYNC_PROTOCOL_VERSION).contains(&protocol)
45}
46
47pub(crate) fn validate_behavior_protocol(protocol: u32) -> Result<()> {
48    if supports_protocol(protocol) {
49        return Ok(());
50    }
51    #[cfg(test)]
52    if protocol == 20 {
53        return Ok(());
54    }
55    Err(SyncCompatibilityError {
56        server_protocol: protocol,
57        client_protocol: SYNC_PROTOCOL_VERSION,
58    }
59    .into())
60}
61
62pub(crate) async fn replica_protocol(conn: &mut SqliteConnection) -> Result<u32> {
63    let protocol = get_meta(conn, "sync_established_protocol")
64        .await?
65        .map(|value| {
66            value
67                .parse::<u32>()
68                .context("invalid established sync protocol")
69        })
70        .transpose()?
71        .unwrap_or(MAINTAINED_PROTOCOL_BASELINE);
72    validate_behavior_protocol(protocol)?;
73    Ok(protocol)
74}
75
76pub(crate) async fn establish_protocol(conn: &mut SqliteConnection, protocol: u32) -> Result<()> {
77    validate_behavior_protocol(protocol)?;
78    set_meta(conn, "sync_established_protocol", &protocol.to_string()).await
79}
80
81/// These literal names are the released baseline, not aliases to extensible domain enums.
82const BASELINE_OPERATIONS: &[&str] = &[
83    "create_task",
84    "move_tasks",
85    "set_field",
86    "resolve_field",
87    "label_add",
88    "label_remove",
89    "note_add",
90    "note_edit",
91    "note_delete",
92    "dependency_add",
93    "dependency_remove",
94    "related_add",
95    "related_remove",
96    "epic_link_add",
97    "epic_link_remove",
98    "attachment_add",
99    "attachment_delete",
100    "create_project",
101    "set_project_metadata",
102    "project_delete",
103    "create_label",
104    "set_label_name",
105    "label_delete",
106    "label_restore",
107    "create_workspace",
108    "set_workspace_field",
109    "create_recurrence_series",
110    "update_recurrence_template",
111    "project_recurrence_occurrence",
112    "resolve_recurrence_occurrence",
113    "set_recurrence_state",
114    "create_metadata_field",
115    "set_metadata_field",
116    "set_task_metadata",
117    "remove_task_metadata",
118    "set_recurrence_metadata",
119    "remove_recurrence_metadata",
120    "open_recurrence_pause",
121    "close_recurrence_pause",
122    "stop_recurrence_series",
123    "publish_device_label",
124];
125/// Operations added by protocol 19. Every supported build reads them; a replica
126/// produces them only once its established protocol reaches 19.
127const PROTOCOL_19_OPERATIONS: &[&str] = &["delete_workspace"];
128const STATUSES: &[&str] = &["inbox", "backlog", "todo", "active", "done", "canceled"];
129const PRIORITIES: &[&str] = &["none", "low", "medium", "high", "urgent"];
130const SOURCES: &[&str] = &["cli", "tui", "api", "ios", "android", "unknown"];
131const TASK_FIELDS: &[&str] = &[
132    "title",
133    "description",
134    "project",
135    "status",
136    "priority",
137    "available_at",
138    "due_on",
139    "deleted",
140    "is_epic",
141];
142
143pub(crate) fn validate_operation(
144    protocol: u32,
145    op: &str,
146    field: Option<&str>,
147    payload: &Value,
148) -> Result<()> {
149    validate_behavior_protocol(protocol)?;
150    let required = required_protocol(op)?;
151    if required > protocol {
152        return Err(SharedOperationCompatibilityError {
153            required_protocol: required,
154            established_protocol: protocol,
155        }
156        .into());
157    }
158    if op == "set_workspace_field" {
159        registered(
160            field.context("error invalid-sync-change field missing")?,
161            &["key", "name"],
162        )?;
163    }
164    if op == "set_metadata_field" {
165        registered(
166            field.context("error invalid-sync-change field missing")?,
167            &["key"],
168        )?;
169    }
170    if op == crate::change_log::op_type::PUBLISH_DEVICE_LABEL {
171        let label = payload
172            .get("label")
173            .and_then(Value::as_str)
174            .context("error invalid-sync-change device-label")?;
175        crate::sync::device_labels::validate_device_label(label)?;
176    }
177    if op == "attachment_add"
178        && let Some(value) = payload.get("media_type").and_then(Value::as_str)
179    {
180        registered(
181            value,
182            &["image/png", "image/jpeg", "image/gif", "image/webp"],
183        )?;
184    }
185    if matches!(op, "set_field" | "resolve_field") {
186        let field = field.context("error invalid-sync-change field missing")?;
187        registered(field, TASK_FIELDS)?;
188        if let Some(value) = payload.get("value").and_then(Value::as_str) {
189            validate_value(field, value)?;
190        }
191    }
192    if matches!(
193        op,
194        "create_task"
195            | "create_recurrence_series"
196            | "project_recurrence_occurrence"
197            | "resolve_recurrence_occurrence"
198            | "set_recurrence_state"
199            | "stop_recurrence_series"
200            | "update_recurrence_template"
201    ) {
202        for key in [
203            "status",
204            "initial_status",
205            "priority",
206            "source",
207            "frequency",
208            "state",
209            "outcome",
210            "due_policy",
211        ] {
212            if let Some(value) = payload.get(key).and_then(Value::as_str) {
213                validate_value(key, value)?;
214            }
215        }
216        if let Some(fields) = payload.get("fields").and_then(Value::as_array) {
217            for pair in fields {
218                if let (Some(field), Some(value)) = (
219                    pair.get(0).and_then(Value::as_str),
220                    pair.get(1).and_then(Value::as_str),
221                ) {
222                    registered(
223                        field,
224                        &[
225                            "title",
226                            "description",
227                            "project",
228                            "priority",
229                            "initial_status",
230                            "available_local_time",
231                            "due_policy",
232                        ],
233                    )?;
234                    validate_value(field, value)?;
235                }
236            }
237        }
238    }
239    Ok(())
240}
241
242pub(crate) fn required_protocol(op: &str) -> Result<u32> {
243    #[cfg(test)]
244    if op == "test_protocol_20" {
245        return Ok(20);
246    }
247    if BASELINE_OPERATIONS.contains(&op) {
248        return Ok(MAINTAINED_PROTOCOL_BASELINE);
249    }
250    if PROTOCOL_19_OPERATIONS.contains(&op) {
251        return Ok(19);
252    }
253    bail!("error invalid-sync-change op_type={op} unregistered-operation");
254}
255
256fn registered(value: &str, values: &[&str]) -> Result<()> {
257    if !values.contains(&value) {
258        bail!("error invalid-sync-change unregistered-value");
259    }
260    Ok(())
261}
262
263fn validate_value(field: &str, value: &str) -> Result<()> {
264    match field {
265        "status" | "initial_status" => registered(value, STATUSES),
266        "priority" => registered(value, PRIORITIES),
267        "source" => {
268            crate::choices::TaskSource::parse(value)?;
269            registered(value, SOURCES)
270        }
271        "frequency" => registered(value, &["daily", "weekly", "monthly", "yearly"]),
272        "state" => registered(value, &["active", "paused", "stopped"]),
273        "outcome" => registered(value, &["completed", "skipped"]),
274        "due_policy" => registered(value, &["none", "same_day"]),
275        _ => Ok(()),
276    }
277}
278
279pub(crate) fn validate_change(protocol: u32, change: &ChangeWire) -> Result<()> {
280    validate_operation(
281        protocol,
282        &change.op_type,
283        change.field.as_deref(),
284        &change.payload,
285    )
286}
287
288#[cfg(test)]
289mod tests {
290    use super::*;
291
292    #[test]
293    fn baseline_vocabulary_is_explicit_and_complete() {
294        let registered = [BASELINE_OPERATIONS, PROTOCOL_19_OPERATIONS].concat();
295        assert_eq!(crate::change_log::op_type::ALL, registered);
296        assert_eq!(crate::choices::STATUSES, STATUSES);
297        assert_eq!(crate::choices::PRIORITIES, PRIORITIES);
298        assert_eq!(crate::choices::TASK_SOURCES, SOURCES);
299        let mut fields = crate::task_fields::TaskField::VERSIONED
300            .map(|field| field.as_str())
301            .to_vec();
302        fields.sort();
303        let mut frozen = TASK_FIELDS.to_vec();
304        frozen.sort();
305        assert_eq!(fields, frozen);
306    }
307
308    #[tokio::test]
309    async fn unsupported_persisted_mode_fails_closed_on_reopen() {
310        let directory = tempfile::tempdir().unwrap();
311        let path = directory.path().join("replica.sqlite");
312        let database = crate::db::Database::open(&path).await.unwrap();
313        let mut conn = database.acquire_writer().await.unwrap();
314        set_meta(&mut conn, "sync_established_protocol", "999")
315            .await
316            .unwrap();
317        drop(conn);
318        drop(database);
319        assert!(crate::db::Database::open(&path).await.is_err());
320    }
321}
aven_core/sync/protocol.rs

aven_core/sync/
protocol.rs
