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 matches!(protocol, 19 | 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];
125const STATUSES: &[&str] = &["inbox", "backlog", "todo", "active", "done", "canceled"];
126const PRIORITIES: &[&str] = &["none", "low", "medium", "high", "urgent"];
127const SOURCES: &[&str] = &["cli", "tui", "api", "ios", "android", "unknown"];
128const TASK_FIELDS: &[&str] = &[
129    "title",
130    "description",
131    "project",
132    "status",
133    "priority",
134    "available_at",
135    "due_on",
136    "deleted",
137    "is_epic",
138];
139
140pub(crate) fn validate_operation(
141    protocol: u32,
142    op: &str,
143    field: Option<&str>,
144    payload: &Value,
145) -> Result<()> {
146    validate_behavior_protocol(protocol)?;
147    let required = required_protocol(op)?;
148    if required > protocol {
149        return Err(SharedOperationCompatibilityError {
150            required_protocol: required,
151            established_protocol: protocol,
152        }
153        .into());
154    }
155    if op == "set_workspace_field" {
156        registered(
157            field.context("error invalid-sync-change field missing")?,
158            &["key", "name"],
159        )?;
160    }
161    if op == "set_metadata_field" {
162        registered(
163            field.context("error invalid-sync-change field missing")?,
164            &["key"],
165        )?;
166    }
167    if op == crate::change_log::op_type::PUBLISH_DEVICE_LABEL {
168        let label = payload
169            .get("label")
170            .and_then(Value::as_str)
171            .context("error invalid-sync-change device-label")?;
172        crate::sync::device_labels::validate_device_label(label)?;
173    }
174    if op == "attachment_add"
175        && let Some(value) = payload.get("media_type").and_then(Value::as_str)
176    {
177        registered(
178            value,
179            &["image/png", "image/jpeg", "image/gif", "image/webp"],
180        )?;
181    }
182    if matches!(op, "set_field" | "resolve_field") {
183        let field = field.context("error invalid-sync-change field missing")?;
184        registered(field, TASK_FIELDS)?;
185        if let Some(value) = payload.get("value").and_then(Value::as_str) {
186            validate_value(field, value)?;
187        }
188    }
189    if matches!(
190        op,
191        "create_task"
192            | "create_recurrence_series"
193            | "project_recurrence_occurrence"
194            | "resolve_recurrence_occurrence"
195            | "set_recurrence_state"
196            | "stop_recurrence_series"
197            | "update_recurrence_template"
198    ) {
199        for key in [
200            "status",
201            "initial_status",
202            "priority",
203            "source",
204            "frequency",
205            "state",
206            "outcome",
207            "due_policy",
208        ] {
209            if let Some(value) = payload.get(key).and_then(Value::as_str) {
210                validate_value(key, value)?;
211            }
212        }
213        if let Some(fields) = payload.get("fields").and_then(Value::as_array) {
214            for pair in fields {
215                if let (Some(field), Some(value)) = (
216                    pair.get(0).and_then(Value::as_str),
217                    pair.get(1).and_then(Value::as_str),
218                ) {
219                    registered(
220                        field,
221                        &[
222                            "title",
223                            "description",
224                            "project",
225                            "priority",
226                            "initial_status",
227                            "available_local_time",
228                            "due_policy",
229                        ],
230                    )?;
231                    validate_value(field, value)?;
232                }
233            }
234        }
235    }
236    Ok(())
237}
238
239pub(crate) fn required_protocol(op: &str) -> Result<u32> {
240    #[cfg(test)]
241    match op {
242        "test_protocol_19" => return Ok(19),
243        "test_protocol_20" => return Ok(20),
244        _ => {}
245    }
246    if !BASELINE_OPERATIONS.contains(&op) {
247        bail!("error invalid-sync-change op_type={op} unregistered-operation");
248    }
249    Ok(MAINTAINED_PROTOCOL_BASELINE)
250}
251
252fn registered(value: &str, values: &[&str]) -> Result<()> {
253    if !values.contains(&value) {
254        bail!("error invalid-sync-change unregistered-value");
255    }
256    Ok(())
257}
258
259fn validate_value(field: &str, value: &str) -> Result<()> {
260    match field {
261        "status" | "initial_status" => registered(value, STATUSES),
262        "priority" => registered(value, PRIORITIES),
263        "source" => {
264            crate::choices::TaskSource::parse(value)?;
265            registered(value, SOURCES)
266        }
267        "frequency" => registered(value, &["daily", "weekly", "monthly", "yearly"]),
268        "state" => registered(value, &["active", "paused", "stopped"]),
269        "outcome" => registered(value, &["completed", "skipped"]),
270        "due_policy" => registered(value, &["none", "same_day"]),
271        _ => Ok(()),
272    }
273}
274
275pub(crate) fn validate_change(protocol: u32, change: &ChangeWire) -> Result<()> {
276    validate_operation(
277        protocol,
278        &change.op_type,
279        change.field.as_deref(),
280        &change.payload,
281    )
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287
288    #[test]
289    fn baseline_vocabulary_is_explicit_and_complete() {
290        assert_eq!(crate::change_log::op_type::ALL, BASELINE_OPERATIONS);
291        assert_eq!(crate::choices::STATUSES, STATUSES);
292        assert_eq!(crate::choices::PRIORITIES, PRIORITIES);
293        assert_eq!(crate::choices::TASK_SOURCES, SOURCES);
294        let mut fields = crate::task_fields::TaskField::VERSIONED
295            .map(|field| field.as_str())
296            .to_vec();
297        fields.sort();
298        let mut frozen = TASK_FIELDS.to_vec();
299        frozen.sort();
300        assert_eq!(fields, frozen);
301    }
302
303    #[tokio::test]
304    async fn unsupported_persisted_mode_fails_closed_on_reopen() {
305        let directory = tempfile::tempdir().unwrap();
306        let path = directory.path().join("replica.sqlite");
307        let database = crate::db::Database::open(&path).await.unwrap();
308        let mut conn = database.acquire_writer().await.unwrap();
309        set_meta(&mut conn, "sync_established_protocol", "999")
310            .await
311            .unwrap();
312        drop(conn);
313        drop(database);
314        assert!(crate::db::Database::open(&path).await.is_err());
315    }
316}
aven_core/sync/protocol.rs

aven_core/sync/
protocol.rs
