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, SyncRequest, SyncResponse};
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    "set_field",
85    "resolve_field",
86    "label_add",
87    "label_remove",
88    "note_add",
89    "note_edit",
90    "note_delete",
91    "dependency_add",
92    "dependency_remove",
93    "related_add",
94    "related_remove",
95    "epic_link_add",
96    "epic_link_remove",
97    "attachment_add",
98    "attachment_delete",
99    "create_project",
100    "set_project_metadata",
101    "project_delete",
102    "create_label",
103    "set_label_name",
104    "label_delete",
105    "label_restore",
106    "create_workspace",
107    "set_workspace_field",
108    "create_recurrence_series",
109    "update_recurrence_template",
110    "project_recurrence_occurrence",
111    "resolve_recurrence_occurrence",
112    "set_recurrence_state",
113    "create_metadata_field",
114    "set_metadata_field",
115    "set_task_metadata",
116    "remove_task_metadata",
117    "set_recurrence_metadata",
118    "remove_recurrence_metadata",
119    "open_recurrence_pause",
120    "close_recurrence_pause",
121    "stop_recurrence_series",
122];
123const STATUSES: &[&str] = &["inbox", "backlog", "todo", "active", "done", "canceled"];
124const PRIORITIES: &[&str] = &["none", "low", "medium", "high", "urgent"];
125const SOURCES: &[&str] = &["cli", "tui", "api", "ios", "android", "unknown"];
126const TASK_FIELDS: &[&str] = &[
127    "title",
128    "description",
129    "project",
130    "status",
131    "priority",
132    "available_at",
133    "due_on",
134    "deleted",
135    "is_epic",
136];
137
138pub(crate) fn validate_operation(
139    protocol: u32,
140    op: &str,
141    field: Option<&str>,
142    payload: &Value,
143) -> Result<()> {
144    validate_behavior_protocol(protocol)?;
145    let required = required_protocol(op)?;
146    if required > protocol {
147        return Err(SharedOperationCompatibilityError {
148            required_protocol: required,
149            established_protocol: protocol,
150        }
151        .into());
152    }
153    if op == "set_workspace_field" {
154        registered(
155            field.context("error invalid-sync-change field missing")?,
156            &["key", "name"],
157        )?;
158    }
159    if op == "set_metadata_field" {
160        registered(
161            field.context("error invalid-sync-change field missing")?,
162            &["key"],
163        )?;
164    }
165    if op == "attachment_add"
166        && let Some(value) = payload.get("media_type").and_then(Value::as_str)
167    {
168        registered(
169            value,
170            &["image/png", "image/jpeg", "image/gif", "image/webp"],
171        )?;
172    }
173    if matches!(op, "set_field" | "resolve_field") {
174        let field = field.context("error invalid-sync-change field missing")?;
175        registered(field, TASK_FIELDS)?;
176        if let Some(value) = payload.get("value").and_then(Value::as_str) {
177            validate_value(field, value)?;
178        }
179    }
180    if matches!(
181        op,
182        "create_task"
183            | "create_recurrence_series"
184            | "project_recurrence_occurrence"
185            | "resolve_recurrence_occurrence"
186            | "set_recurrence_state"
187            | "stop_recurrence_series"
188            | "update_recurrence_template"
189    ) {
190        for key in [
191            "status",
192            "initial_status",
193            "priority",
194            "source",
195            "frequency",
196            "state",
197            "outcome",
198            "due_policy",
199        ] {
200            if let Some(value) = payload.get(key).and_then(Value::as_str) {
201                validate_value(key, value)?;
202            }
203        }
204        if let Some(fields) = payload.get("fields").and_then(Value::as_array) {
205            for pair in fields {
206                if let (Some(field), Some(value)) = (
207                    pair.get(0).and_then(Value::as_str),
208                    pair.get(1).and_then(Value::as_str),
209                ) {
210                    registered(
211                        field,
212                        &[
213                            "title",
214                            "description",
215                            "project",
216                            "priority",
217                            "initial_status",
218                            "available_local_time",
219                            "due_policy",
220                        ],
221                    )?;
222                    validate_value(field, value)?;
223                }
224            }
225        }
226    }
227    Ok(())
228}
229
230fn required_protocol(op: &str) -> Result<u32> {
231    #[cfg(test)]
232    match op {
233        "test_protocol_19" => return Ok(19),
234        "test_protocol_20" => return Ok(20),
235        _ => {}
236    }
237    if !BASELINE_OPERATIONS.contains(&op) {
238        bail!("error invalid-sync-change op_type={op} unregistered-operation");
239    }
240    Ok(MAINTAINED_PROTOCOL_BASELINE)
241}
242
243fn registered(value: &str, values: &[&str]) -> Result<()> {
244    if !values.contains(&value) {
245        bail!("error invalid-sync-change unregistered-value");
246    }
247    Ok(())
248}
249
250fn validate_value(field: &str, value: &str) -> Result<()> {
251    match field {
252        "status" | "initial_status" => registered(value, STATUSES),
253        "priority" => registered(value, PRIORITIES),
254        "source" => {
255            crate::choices::TaskSource::parse(value)?;
256            registered(value, SOURCES)
257        }
258        "frequency" => registered(value, &["daily", "weekly", "monthly", "yearly"]),
259        "state" => registered(value, &["active", "paused", "stopped"]),
260        "outcome" => registered(value, &["completed", "skipped"]),
261        "due_policy" => registered(value, &["none", "same_day"]),
262        _ => Ok(()),
263    }
264}
265
266pub(crate) fn validate_change(protocol: u32, change: &ChangeWire) -> Result<()> {
267    validate_operation(
268        protocol,
269        &change.op_type,
270        change.field.as_deref(),
271        &change.payload,
272    )
273}
274
275pub fn discovery_request(protocol: u32, client_id: String) -> SyncRequest {
276    SyncRequest {
277        protocol_version: Some(protocol),
278        client_id,
279        after: i64::MAX,
280        pull_limit: Some(1),
281        changes: Vec::new(),
282    }
283}
284
285pub fn validate_discovery_response(protocol: u32, response: &SyncResponse) -> Result<()> {
286    super::wire::validate_response_at_protocol(protocol, i64::MAX, 1, &[], response)
287}
288
289pub fn protocol_mismatch(detail: &str) -> Option<(u32, u32)> {
290    let mut fields = detail
291        .trim()
292        .strip_prefix("error sync-protocol-unsupported ")?
293        .split_whitespace();
294    let client = fields.next()?.strip_prefix("client=")?.parse().ok()?;
295    let server = fields.next()?.strip_prefix("server=")?.parse().ok()?;
296    if fields.next().is_some() || client == server {
297        return None;
298    }
299    Some((client, server))
300}
301
302#[cfg(test)]
303mod tests {
304    use super::*;
305
306    #[test]
307    fn baseline_vocabulary_is_explicit_and_complete() {
308        let source = include_str!("../change_log.rs");
309        let operations = source
310            .lines()
311            .filter_map(|line| {
312                let (_, value) = line
313                    .trim()
314                    .strip_prefix("pub const ")?
315                    .split_once(": &str = \"")?;
316                value.strip_suffix("\";")
317            })
318            .collect::<Vec<_>>();
319        assert_eq!(operations, BASELINE_OPERATIONS);
320        assert_eq!(crate::choices::STATUSES, STATUSES);
321        assert_eq!(crate::choices::PRIORITIES, PRIORITIES);
322        assert_eq!(crate::choices::TASK_SOURCES, SOURCES);
323        let mut fields = crate::task_fields::TaskField::VERSIONED
324            .map(|field| field.as_str())
325            .to_vec();
326        fields.sort();
327        let mut frozen = TASK_FIELDS.to_vec();
328        frozen.sort();
329        assert_eq!(fields, frozen);
330    }
331
332    #[tokio::test]
333    async fn unsupported_persisted_mode_fails_closed_on_reopen() {
334        let directory = tempfile::tempdir().unwrap();
335        let path = directory.path().join("replica.sqlite");
336        let database = crate::db::Database::open(&path).await.unwrap();
337        let mut conn = database.acquire_writer().await.unwrap();
338        set_meta(&mut conn, "sync_established_protocol", "999")
339            .await
340            .unwrap();
341        drop(conn);
342        drop(database);
343        assert!(crate::db::Database::open(&path).await.is_err());
344    }
345}
aven_core/sync/protocol.rs

aven_core/sync/
protocol.rs
