aven_core/sync/
wire.rs

1use std::collections::{HashMap, HashSet};
2
3use anyhow::{Context, Result, bail};
4use chrono::{NaiveDate, NaiveTime};
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7
8use crate::attachments::validation::{
9    MAX_BLOB_BYTES, validate_alt_text, validate_blob_size, validate_dimensions, validate_filename,
10    validate_media_type,
11};
12use crate::change_log::op_type;
13use crate::choices::TaskSource;
14use crate::ids::{BASE32, ProjectId, WorkspaceId};
15use crate::recurrence::{
16    RecurrenceDuePolicy, RecurrenceFrequency, RecurrenceOutcome, RecurrenceRule,
17    RecurrenceSchedule, RecurrenceSeriesId, RecurrenceSeriesState, TimeZoneId, WeekdaySet,
18    derive_occurrence_identity, is_slot, next_slot_after,
19};
20use crate::task_fields::TaskField;
21
22pub const SYNC_PROTOCOL_VERSION: u32 = 12;
23const MAX_CHANGE_PAYLOAD_BYTES: usize = 64 * 1024;
24pub fn sync_server_url_is_valid(server: &str) -> bool {
25    let Ok(url) = url::Url::parse(server) else {
26        return false;
27    };
28    matches!(url.scheme(), "http" | "https")
29        && url.host_str().is_some()
30        && url.username().is_empty()
31        && url.password().is_none()
32        && url.query().is_none()
33        && url.fragment().is_none()
34}
35pub const MAX_PUSH_BATCH: usize = 256;
36pub const MAX_PULL_BATCH: u32 = 512;
37pub const MAX_BLOB_TRANSFER_OBJECTS: usize = 16;
38pub const MAX_BLOB_TRANSFER_BYTES: u64 = 64 * 1024 * 1024;
39pub const DAEMON_SYNC_PAGE_BUDGET: usize = 8;
40pub const DAEMON_INCOMPLETE_RESCHEDULE_MS: u64 = 100;
41
42#[derive(Debug, Clone, Serialize, Deserialize)]
43pub struct ChangeWire {
44    pub change_id: String,
45    pub client_id: String,
46    pub local_seq: i64,
47    pub entity_type: String,
48    pub entity_id: String,
49    pub field: Option<String>,
50    pub op_type: String,
51    pub payload: Value,
52    pub base_version: Option<String>,
53    pub created_at: String,
54    pub server_seq: Option<i64>,
55}
56
57#[derive(Debug, Clone, Serialize, Deserialize)]
58pub struct PushAck {
59    pub change_id: String,
60    pub server_seq: i64,
61}
62
63#[derive(Debug)]
64pub struct ChangeRow {
65    pub change_id: String,
66    pub client_id: String,
67    pub local_seq: i64,
68    pub entity_type: String,
69    pub entity_id: String,
70    pub field: Option<String>,
71    pub op_type: String,
72    pub payload: String,
73    pub base_version: Option<String>,
74    pub created_at: String,
75    pub server_seq: Option<i64>,
76}
77
78impl ChangeRow {
79    pub fn into_wire(self) -> ChangeWire {
80        ChangeWire {
81            change_id: self.change_id,
82            client_id: self.client_id,
83            local_seq: self.local_seq,
84            entity_type: self.entity_type,
85            entity_id: self.entity_id,
86            field: self.field,
87            op_type: self.op_type,
88            payload: serde_json::from_str(&self.payload).unwrap_or(Value::Null),
89            base_version: self.base_version,
90            created_at: self.created_at,
91            server_seq: self.server_seq,
92        }
93    }
94}
95
96#[derive(Debug, Clone, Serialize, Deserialize)]
97pub struct SyncRequest {
98    #[serde(default)]
99    pub protocol_version: Option<u32>,
100    pub client_id: String,
101    pub after: i64,
102    #[serde(default)]
103    pub pull_limit: Option<u32>,
104    pub changes: Vec<ChangeWire>,
105}
106
107#[derive(Debug, Serialize, Deserialize)]
108pub struct SyncResponse {
109    pub protocol_version: u32,
110    pub cursor: i64,
111    pub has_more: bool,
112    #[serde(default)]
113    pub push_acks: Vec<PushAck>,
114    pub changes: Vec<ChangeWire>,
115}
116
117#[derive(Debug, Clone, Serialize, Deserialize)]
118pub struct BlobUploadContract {
119    pub workspace_id: String,
120    pub sha256: String,
121    pub byte_size: i64,
122    pub media_type: String,
123    pub width: i64,
124    pub height: i64,
125}
126
127#[derive(Debug, Serialize, Deserialize)]
128pub struct MissingBlobsRequest {
129    pub blobs: Vec<BlobUploadContract>,
130}
131
132#[derive(Debug, Serialize, Deserialize)]
133pub struct MissingBlobsResponse {
134    pub missing: Vec<String>,
135}
136
137pub fn validate_blob_contracts(blobs: &[BlobUploadContract]) -> Result<()> {
138    if blobs.len() > MAX_PUSH_BATCH {
139        bail!(
140            "error blob-batch-too-large limit={} got={}",
141            MAX_PUSH_BATCH,
142            blobs.len()
143        );
144    }
145    let mut seen = HashSet::with_capacity(blobs.len());
146    let mut hashes = HashSet::with_capacity(blobs.len());
147    for blob in blobs {
148        ensure_sync_id("workspace_id", &blob.workspace_id)?;
149        validate_sha256_for_sync(&blob.sha256)?;
150        validate_blob_size_for_sync(blob.byte_size)?;
151        map_attachment_validation(validate_media_type(&blob.media_type))?;
152        map_attachment_validation(validate_dimensions(Some(blob.width), Some(blob.height)))?;
153        if !seen.insert((blob.workspace_id.as_str(), blob.sha256.as_str())) {
154            bail!("error duplicate-blob-contract");
155        }
156        hashes.insert(blob.sha256.as_str());
157    }
158    if hashes.len() > MAX_BLOB_TRANSFER_OBJECTS {
159        bail!(
160            "error blob-batch-too-large limit={} got={}",
161            MAX_BLOB_TRANSFER_OBJECTS,
162            hashes.len()
163        );
164    }
165    Ok(())
166}
167
168pub fn validate_blob_hashes(hashes: &[String]) -> Result<()> {
169    if hashes.len() > MAX_BLOB_TRANSFER_OBJECTS {
170        bail!(
171            "error blob-batch-too-large limit={} got={}",
172            MAX_BLOB_TRANSFER_OBJECTS,
173            hashes.len()
174        );
175    }
176    let mut seen = HashSet::with_capacity(hashes.len());
177    for hash in hashes {
178        validate_sha256_for_sync(hash)?;
179        if !seen.insert(hash.as_str()) {
180            bail!("error duplicate-blob-hash");
181        }
182    }
183    Ok(())
184}
185
186#[derive(Debug)]
187struct SyncProtocolError {
188    client: u32,
189    server: u32,
190}
191
192impl std::fmt::Display for SyncProtocolError {
193    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
194        write!(
195            f,
196            "error sync-protocol-unsupported client={} server={}",
197            self.client, self.server
198        )
199    }
200}
201
202impl std::error::Error for SyncProtocolError {}
203
204fn sync_protocol_error(client: u32, server: u32) -> anyhow::Error {
205    anyhow::Error::new(SyncProtocolError { client, server })
206}
207
208pub fn validate_sync_protocol_version(client: u32, server: u32) -> Result<()> {
209    if client != server {
210        return Err(sync_protocol_error(client, server));
211    }
212    Ok(())
213}
214
215pub fn validate_sync_request_protocol_version(client: Option<u32>) -> Result<()> {
216    validate_sync_protocol_version(client.unwrap_or(0), SYNC_PROTOCOL_VERSION)
217}
218
219pub fn request_pull_limit(requested: Option<u32>) -> Result<u32> {
220    match requested {
221        None => Ok(MAX_PULL_BATCH),
222        Some(limit @ 1..=MAX_PULL_BATCH) => Ok(limit),
223        Some(limit) => {
224            bail!("error sync-pull-limit-out-of-range min=1 max={MAX_PULL_BATCH} got={limit}")
225        }
226    }
227}
228
229#[derive(Debug, Clone, Copy)]
230pub struct ValidatedSyncRequestEnvelope {
231    pub after: i64,
232    pub pull_limit: u32,
233    pub push_count: usize,
234}
235
236pub fn validate_sync_request_envelope(
237    request: &SyncRequest,
238) -> Result<ValidatedSyncRequestEnvelope> {
239    validate_sync_request_protocol_version(request.protocol_version)?;
240    validate_request_cursor(request.after)?;
241    validate_push_batch_size(request.changes.len())?;
242    Ok(ValidatedSyncRequestEnvelope {
243        after: request.after,
244        pull_limit: request_pull_limit(request.pull_limit)?,
245        push_count: request.changes.len(),
246    })
247}
248
249fn validate_request_cursor(after: i64) -> Result<()> {
250    if after < 0 {
251        bail!("error sync-after-out-of-range min=0 got={after}");
252    }
253    Ok(())
254}
255
256fn validate_push_batch_size(len: usize) -> Result<()> {
257    if len > MAX_PUSH_BATCH {
258        bail!("error sync-push-too-large limit={MAX_PUSH_BATCH} got={len}");
259    }
260    Ok(())
261}
262
263pub fn validate_sync_response_for_request(
264    after: i64,
265    pull_limit: u32,
266    request_change_ids: &[String],
267    response: &SyncResponse,
268) -> Result<()> {
269    validate_sync_protocol_version(SYNC_PROTOCOL_VERSION, response.protocol_version)?;
270    if response.changes.len() > pull_limit as usize {
271        bail!(
272            "error invalid-sync-response pull-too-large limit={} got={}",
273            pull_limit,
274            response.changes.len()
275        );
276    }
277    if response.cursor < after {
278        bail!(
279            "error invalid-sync-response cursor-regressed after={} cursor={}",
280            after,
281            response.cursor
282        );
283    }
284    validate_push_acks(request_change_ids, response)?;
285    validate_pull_page(after, pull_limit, response)?;
286    validate_push_pull_overlap(response)?;
287    Ok(())
288}
289
290#[derive(Debug, Clone, Copy, PartialEq, Eq)]
291enum ChangeDirection {
292    Pushed,
293    Pulled,
294}
295
296pub fn validate_pushed_change(change: &ChangeWire) -> Result<()> {
297    validate_change_shape(change, ChangeDirection::Pushed)
298}
299
300fn validate_pulled_change(change: &ChangeWire) -> Result<()> {
301    validate_change_shape(change, ChangeDirection::Pulled)
302}
303
304fn validate_change_shape(change: &ChangeWire, direction: ChangeDirection) -> Result<()> {
305    ensure_non_empty("change_id", &change.change_id)?;
306    ensure_non_empty("client_id", &change.client_id)?;
307    ensure_non_empty("entity_id", &change.entity_id)?;
308    ensure_non_empty("op_type", &change.op_type)?;
309    ensure_non_empty("entity_type", &change.entity_type)?;
310    if direction == ChangeDirection::Pushed {
311        ensure_sync_id("change_id", &change.change_id)?;
312    }
313    validate_change_server_seq(change, direction)?;
314    if !change.payload.is_object() {
315        bail!("error invalid-sync-change payload expected-object");
316    }
317    if serde_json::to_vec(&change.payload)?.len() > MAX_CHANGE_PAYLOAD_BYTES {
318        bail!("error invalid-sync-change payload-too-large limit={MAX_CHANGE_PAYLOAD_BYTES}");
319    }
320
321    match change.op_type.as_str() {
322        op_type::CREATE_WORKSPACE => {
323            ensure_entity_type(change, "workspace")?;
324            ensure_sync_id("entity_id", &change.entity_id)?;
325            required_string_payload("key", &change.payload)?;
326            required_string_payload("name", &change.payload)?;
327            required_string_payload("created_at", &change.payload)?;
328        }
329        op_type::SET_WORKSPACE_FIELD => {
330            ensure_entity_type(change, "workspace")?;
331            ensure_sync_id("entity_id", &change.entity_id)?;
332            let field = change
333                .field
334                .as_deref()
335                .filter(|field| !field.trim().is_empty())
336                .context("error invalid-sync-change field missing")?;
337            if !matches!(field, "name" | "key") {
338                bail!("error invalid-sync-change field={field}");
339            }
340            required_string_payload("value", &change.payload)?;
341        }
342        op_type::CREATE_PROJECT => {
343            ensure_entity_type(change, "project")?;
344            ensure_project_id("entity_id", &change.entity_id)?;
345            optional_workspace_payload(&change.payload)?;
346            required_string_payload("key", &change.payload)?;
347            required_string_payload("name", &change.payload)?;
348            required_string_payload("prefix", &change.payload)?;
349            required_string_payload("created_at", &change.payload)?;
350        }
351        op_type::SET_PROJECT_METADATA => {
352            ensure_entity_type(change, "project")?;
353            ensure_project_id("entity_id", &change.entity_id)?;
354            optional_workspace_payload(&change.payload)?;
355            required_string_payload("key", &change.payload)?;
356            required_string_payload("name", &change.payload)?;
357            required_string_payload("prefix", &change.payload)?;
358            required_string_payload("updated_at", &change.payload)?;
359        }
360        op_type::CREATE_LABEL => {
361            ensure_entity_type(change, "label")?;
362            optional_workspace_payload(&change.payload)?;
363            required_string_payload("name", &change.payload)?;
364            required_string_payload("created_at", &change.payload)?;
365        }
366        op_type::CREATE_TASK => {
367            ensure_entity_type(change, "task")?;
368            ensure_sync_id("entity_id", &change.entity_id)?;
369            optional_workspace_payload(&change.payload)?;
370            required_string_payload("title", &change.payload)?;
371            let project_id = required_string_payload("project_id", &change.payload)?;
372            ensure_project_id("project_id", &project_id)?;
373            optional_string_payload("description", &change.payload)?;
374            let recurrence_task = change
375                .payload
376                .get("series_id")
377                .and_then(Value::as_str)
378                .is_some();
379            if recurrence_task {
380                let series_id = required_string_payload("series_id", &change.payload)?;
381                series_id.parse::<RecurrenceSeriesId>().map_err(|_| {
382                    anyhow::anyhow!("error invalid-sync-change series_id invalid-id")
383                })?;
384                validate_recurrence_task(change)?;
385            } else {
386                required_string_payload("project_key", &change.payload)?;
387                required_string_payload("project_name", &change.payload)?;
388                required_string_payload("project_prefix", &change.payload)?;
389            }
390            if let Some(status) = optional_string_payload("status", &change.payload)? {
391                validate_sync_task_field_value(TaskField::Status, &status)?;
392            }
393            if let Some(priority) = optional_string_payload("priority", &change.payload)? {
394                validate_sync_task_field_value(TaskField::Priority, &priority)?;
395            }
396            if let Some(source) = optional_string_payload("source", &change.payload)? {
397                TaskSource::parse(&source)
398                    .context("error invalid-sync-change invalid-task-source")?;
399            }
400            if let Some(available_at) = optional_string_payload("available_at", &change.payload)? {
401                validate_sync_task_field_value(TaskField::AvailableAt, &available_at)?;
402            }
403            if let Some(due_on) = optional_string_payload("due_on", &change.payload)? {
404                validate_sync_task_field_value(TaskField::DueOn, &due_on)?;
405            }
406            if let Some(is_epic) = optional_string_payload("is_epic", &change.payload)? {
407                validate_sync_task_field_value(TaskField::IsEpic, &is_epic)?;
408            }
409            optional_string_array_payload("labels", &change.payload)?;
410            optional_string_payload("created_at", &change.payload)?;
411        }
412        op_type::SET_FIELD | op_type::RESOLVE_FIELD => {
413            ensure_entity_type(change, "task")?;
414            ensure_sync_id("entity_id", &change.entity_id)?;
415            optional_workspace_payload(&change.payload)?;
416            let field = change
417                .field
418                .as_deref()
419                .filter(|field| !field.trim().is_empty())
420                .context("error invalid-sync-change field missing")?;
421            let task_field = TaskField::parse_for_sync(field)?;
422            let value = required_string_payload("value", &change.payload)?;
423            validate_sync_task_field_value(task_field, &value)?;
424            if task_field == TaskField::Project {
425                let project_id = required_string_payload("project_id", &change.payload)?;
426                ensure_project_id("project_id", &project_id)?;
427                if value != project_id {
428                    bail!("error invalid-sync-change project-value-mismatch");
429                }
430                required_string_payload("project_key", &change.payload)?;
431                required_string_payload("project_name", &change.payload)?;
432                required_string_payload("project_prefix", &change.payload)?;
433            }
434        }
435        op_type::LABEL_ADD | op_type::LABEL_REMOVE => {
436            ensure_entity_type(change, "task")?;
437            ensure_sync_id("entity_id", &change.entity_id)?;
438            optional_workspace_payload(&change.payload)?;
439            required_string_payload("label", &change.payload)?;
440        }
441        op_type::NOTE_ADD => {
442            ensure_entity_type(change, "task")?;
443            ensure_sync_id("entity_id", &change.entity_id)?;
444            optional_workspace_payload(&change.payload)?;
445            let note_id = required_string_payload("note_id", &change.payload)?;
446            ensure_sync_id("note_id", &note_id)?;
447            required_string_payload("body", &change.payload)?;
448            required_string_payload("created_at", &change.payload)?;
449        }
450        op_type::DEPENDENCY_ADD | op_type::DEPENDENCY_REMOVE => {
451            ensure_entity_type(change, "task")?;
452            ensure_sync_id("entity_id", &change.entity_id)?;
453            required_workspace_payload(&change.payload)?;
454            let depends_on_task_id =
455                required_string_payload("depends_on_task_id", &change.payload)?;
456            ensure_sync_id("depends_on_task_id", &depends_on_task_id)?;
457            if change.entity_id == depends_on_task_id {
458                bail!("error invalid-sync-change dependency-self");
459            }
460        }
461        op_type::EPIC_LINK_ADD | op_type::EPIC_LINK_REMOVE => {
462            ensure_entity_type(change, "task")?;
463            ensure_sync_id("entity_id", &change.entity_id)?;
464            required_workspace_payload(&change.payload)?;
465            let epic_task_id = required_string_payload("epic_task_id", &change.payload)?;
466            ensure_sync_id("epic_task_id", &epic_task_id)?;
467            if change.entity_id == epic_task_id {
468                bail!("error invalid-sync-change epic-self");
469            }
470            if change.op_type == op_type::EPIC_LINK_ADD {
471                required_timestamp_payload("created_at", &change.payload)?;
472            }
473        }
474        op_type::PROJECT_DELETE => {
475            ensure_entity_type(change, "project")?;
476            ensure_project_id("entity_id", &change.entity_id)?;
477            required_workspace_payload(&change.payload)?;
478            required_timestamp_payload("deleted_at", &change.payload)?;
479        }
480        op_type::SET_LABEL_NAME => {
481            ensure_entity_type(change, "label")?;
482            required_workspace_payload(&change.payload)?;
483            let name = required_string_payload("name", &change.payload)?;
484            if name != change.entity_id {
485                bail!("error invalid-sync-change label-value-mismatch");
486            }
487            required_string_payload("new_name", &change.payload)?;
488            required_timestamp_payload("renamed_at", &change.payload)?;
489        }
490        op_type::LABEL_DELETE => {
491            ensure_entity_type(change, "label")?;
492            required_workspace_payload(&change.payload)?;
493            let name = required_string_payload("name", &change.payload)?;
494            if name != change.entity_id {
495                bail!("error invalid-sync-change label-value-mismatch");
496            }
497            required_timestamp_payload("deleted_at", &change.payload)?;
498        }
499        op_type::NOTE_EDIT => {
500            ensure_entity_type(change, "task")?;
501            ensure_sync_id("entity_id", &change.entity_id)?;
502            if change.field.as_deref() != Some("notes") {
503                bail!("error invalid-sync-change field=notes");
504            }
505            required_workspace_payload(&change.payload)?;
506            let note_id = required_string_payload("note_id", &change.payload)?;
507            ensure_sync_id("note_id", &note_id)?;
508            required_string_payload("body", &change.payload)?;
509            required_timestamp_payload("edited_at", &change.payload)?;
510        }
511        op_type::LABEL_RESTORE => {
512            ensure_entity_type(change, "label")?;
513            required_workspace_payload(&change.payload)?;
514            let name = required_string_payload("name", &change.payload)?;
515            if name != change.entity_id {
516                bail!("error invalid-sync-change label-value-mismatch");
517            }
518            required_timestamp_payload("created_at", &change.payload)?;
519            required_timestamp_payload("restored_at", &change.payload)?;
520            for key in ["task_ids", "series_ids"] {
521                optional_string_array_payload(key, &change.payload)?;
522                if change.payload.get(key).is_none() {
523                    bail!("error invalid-sync-change payload.{key} missing");
524                }
525            }
526            for task_id in string_array_payload("task_ids", &change.payload)? {
527                ensure_sync_id("task_ids", &task_id)?;
528            }
529            for series_id in string_array_payload("series_ids", &change.payload)? {
530                ensure_sync_id("series_ids", &series_id)?;
531            }
532        }
533        op_type::NOTE_DELETE => {
534            ensure_entity_type(change, "task")?;
535            ensure_sync_id("entity_id", &change.entity_id)?;
536            if change.field.as_deref() != Some("notes") {
537                bail!("error invalid-sync-change field=notes");
538            }
539            required_workspace_payload(&change.payload)?;
540            let note_id = required_string_payload("note_id", &change.payload)?;
541            ensure_sync_id("note_id", &note_id)?;
542            required_timestamp_payload("deleted_at", &change.payload)?;
543        }
544        op_type::CREATE_RECURRENCE_SERIES => validate_recurrence_create(change)?,
545        op_type::UPDATE_RECURRENCE_TEMPLATE => validate_recurrence_template(change)?,
546        op_type::PROJECT_RECURRENCE_OCCURRENCE => validate_recurrence_projection(change)?,
547        op_type::RESOLVE_RECURRENCE_OCCURRENCE => validate_recurrence_outcome(change)?,
548        op_type::SET_RECURRENCE_STATE | op_type::STOP_RECURRENCE_SERIES => {
549            validate_recurrence_state(change)?
550        }
551        op_type::OPEN_RECURRENCE_PAUSE => validate_recurrence_pause(change, true)?,
552        op_type::CLOSE_RECURRENCE_PAUSE => validate_recurrence_pause(change, false)?,
553        op_type::ATTACHMENT_ADD => validate_attachment_add_change(change)?,
554        op_type::ATTACHMENT_DELETE => validate_attachment_delete_change(change)?,
555        _ => bail!("error invalid-sync-change op_type={}", change.op_type),
556    }
557    Ok(())
558}
559
560fn validate_recurrence_task(change: &ChangeWire) -> Result<()> {
561    let workspace_id: WorkspaceId = required_string_payload("workspace_id", &change.payload)?
562        .parse()
563        .map_err(|_| anyhow::anyhow!("error invalid-sync-change workspace_id invalid-id"))?;
564    let series_id: RecurrenceSeriesId = required_string_payload("series_id", &change.payload)?
565        .parse()
566        .map_err(|_| anyhow::anyhow!("error invalid-sync-change series_id invalid-id"))?;
567    let slot_on = recurrence_date_payload("slot_on", &change.payload)?;
568    let schedule = recurrence_schedule_payload(&change.payload)?;
569    if !is_slot(&schedule.rule, schedule.start_on, slot_on) {
570        bail!("error invalid-sync-change recurrence-slot-off-lattice");
571    }
572    let identity = derive_occurrence_identity(&workspace_id, &series_id, &schedule, slot_on)
573        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
574    let slot = crate::recurrence::slot_values(&schedule, slot_on)
575        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
576    for (key, expected) in [
577        ("task_id", identity.task_id.as_str()),
578        ("created_at", identity.created_at.as_str()),
579        ("updated_at", identity.updated_at.as_str()),
580        ("available_at", slot.available_at.as_str()),
581        ("due_on", slot.due_on.as_deref().unwrap_or("")),
582        ("task_change_id", identity.task_change_id.as_str()),
583        (
584            "occurrence_change_id",
585            identity.occurrence_change_id.as_str(),
586        ),
587        (
588            "task_field_version_seed",
589            identity.field_version_seeds.task.as_str(),
590        ),
591        (
592            "occurrence_field_version_seed",
593            identity.field_version_seeds.occurrence.as_str(),
594        ),
595    ] {
596        if required_string_payload(key, &change.payload)? != expected {
597            bail!("error invalid-sync-change recurrence-deterministic-mismatch field={key}");
598        }
599    }
600    if change.entity_id != identity.task_id.as_str()
601        || change.change_id != identity.task_change_id
602        || change.created_at != identity.created_at
603    {
604        bail!("error invalid-sync-change recurrence-deterministic-mismatch field=change_identity");
605    }
606    Ok(())
607}
608
609fn validate_recurrence_create(change: &ChangeWire) -> Result<()> {
610    ensure_entity_type(change, "recurrence_series")?;
611    let series_id: RecurrenceSeriesId = change
612        .entity_id
613        .parse()
614        .map_err(|_| anyhow::anyhow!("error invalid-sync-change entity_id invalid-id"))?;
615    required_workspace_payload(&change.payload)?;
616    if required_string_payload("series_id", &change.payload)? != series_id.as_str() {
617        bail!("error invalid-sync-change recurrence-series-id-mismatch");
618    }
619    required_string_payload("title", &change.payload)?;
620    required_string_payload("description", &change.payload)?;
621    let project_id = required_string_payload("project_id", &change.payload)?;
622    ensure_project_id("project_id", &project_id)?;
623    required_string_payload("project_key", &change.payload)?;
624    required_string_payload("project_name", &change.payload)?;
625    required_string_payload("project_prefix", &change.payload)?;
626    validate_sync_task_field_value(
627        TaskField::Priority,
628        &required_string_payload("priority", &change.payload)?,
629    )?;
630    let status = required_string_payload("initial_status", &change.payload)?;
631    validate_sync_task_field_value(TaskField::Status, &status)?;
632    if matches!(status.as_str(), "done" | "canceled") {
633        bail!("error invalid-sync-change recurrence-terminal-template");
634    }
635    recurrence_schedule_payload(&change.payload)?;
636    optional_string_array_payload("labels", &change.payload)?;
637    if required_string_payload("state", &change.payload)? != "active"
638        || !required_string_payload("stopped_at", &change.payload)?.is_empty()
639    {
640        bail!("error invalid-sync-change recurrence-create-state");
641    }
642    required_timestamp_payload("created_at", &change.payload)?;
643    required_timestamp_payload("updated_at", &change.payload)?;
644    Ok(())
645}
646
647fn validate_recurrence_template(change: &ChangeWire) -> Result<()> {
648    validate_recurrence_entity(change)?;
649    let fields = change
650        .payload
651        .get("fields")
652        .and_then(Value::as_array)
653        .context("error invalid-sync-change payload.fields missing")?;
654    if fields.len() > 8 {
655        bail!("error invalid-sync-change recurrence-template-fields-too-large");
656    }
657    let base_versions = change
658        .payload
659        .get("base_versions")
660        .and_then(Value::as_object)
661        .context("error invalid-sync-change payload.base_versions missing")?;
662    let mut seen = HashSet::new();
663    for pair in fields {
664        let pair = pair
665            .as_array()
666            .filter(|pair| pair.len() == 2)
667            .context("error invalid-sync-change recurrence-template-field-pair")?;
668        let field = pair[0]
669            .as_str()
670            .context("error invalid-sync-change recurrence-template-field")?;
671        let value = pair[1]
672            .as_str()
673            .context("error invalid-sync-change recurrence-template-value")?;
674        if !seen.insert(field) {
675            bail!("error invalid-sync-change recurrence-template-duplicate-field");
676        }
677        if !matches!(
678            base_versions.get(field),
679            Some(Value::String(_)) | Some(Value::Null)
680        ) {
681            bail!("error invalid-sync-change recurrence-template-base-version field={field}");
682        }
683        match field {
684            "title" | "description" => {}
685            "project" => ensure_project_id("project_id", value)?,
686            "priority" => validate_sync_task_field_value(TaskField::Priority, value)?,
687            "initial_status" => {
688                validate_sync_task_field_value(TaskField::Status, value)?;
689                if matches!(value, "done" | "canceled") {
690                    bail!("error invalid-sync-change recurrence-terminal-template");
691                }
692            }
693            "available_local_time" => {
694                validate_local_time(value)?;
695            }
696            "due_policy" => {
697                RecurrenceDuePolicy::parse(value)
698                    .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
699            }
700            _ => bail!("error invalid-sync-change recurrence-template-field={field}"),
701        }
702    }
703    optional_string_array_payload("labels", &change.payload)?;
704    let labels_changed = change
705        .payload
706        .get("labels_changed")
707        .and_then(Value::as_bool)
708        .context("error invalid-sync-change payload.labels_changed missing")?;
709    if labels_changed
710        && !matches!(
711            base_versions.get("labels"),
712            Some(Value::String(_)) | Some(Value::Null)
713        )
714    {
715        bail!("error invalid-sync-change recurrence-template-base-version field=labels");
716    }
717    required_timestamp_payload("updated_at", &change.payload)?;
718    Ok(())
719}
720
721fn validate_recurrence_projection(change: &ChangeWire) -> Result<()> {
722    validate_recurrence_entity(change)?;
723    if change.field.as_deref() != Some("projection") {
724        bail!("error invalid-sync-change field=projection");
725    }
726    let workspace_id: WorkspaceId = required_string_payload("workspace_id", &change.payload)?
727        .parse()
728        .map_err(|_| anyhow::anyhow!("error invalid-sync-change workspace_id invalid-id"))?;
729    let series_id: RecurrenceSeriesId = change.entity_id.parse().unwrap();
730    let slot_on = recurrence_date_payload("slot_on", &change.payload)?;
731    let schedule = recurrence_schedule_payload(&change.payload)?;
732    if !is_slot(&schedule.rule, schedule.start_on, slot_on) {
733        bail!("error invalid-sync-change recurrence-slot-off-lattice");
734    }
735    let identity = derive_occurrence_identity(&workspace_id, &series_id, &schedule, slot_on)
736        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
737    for (key, expected) in [
738        ("task_id", identity.task_id.as_str()),
739        (
740            "projected_at",
741            identity.occurrence_link.projected_at.as_str(),
742        ),
743        ("task_change_id", identity.task_change_id.as_str()),
744        (
745            "occurrence_change_id",
746            identity.occurrence_change_id.as_str(),
747        ),
748        (
749            "task_field_version_seed",
750            identity.field_version_seeds.task.as_str(),
751        ),
752        (
753            "occurrence_field_version_seed",
754            identity.field_version_seeds.occurrence.as_str(),
755        ),
756    ] {
757        if required_string_payload(key, &change.payload)? != expected {
758            bail!("error invalid-sync-change recurrence-deterministic-mismatch field={key}");
759        }
760    }
761    if change.change_id != identity.occurrence_change_id
762        || change.created_at != identity.occurrence_link.projected_at
763    {
764        bail!("error invalid-sync-change recurrence-deterministic-mismatch field=change_id");
765    }
766    Ok(())
767}
768
769fn validate_recurrence_outcome(change: &ChangeWire) -> Result<()> {
770    validate_recurrence_entity(change)?;
771    validate_timestamp_value("created_at", &change.created_at)?;
772    if change.field.as_deref() != Some("outcome") {
773        bail!("error invalid-sync-change field=outcome");
774    }
775    let slot_on = recurrence_date_payload("slot_on", &change.payload)?;
776    let schedule = recurrence_schedule_payload(&change.payload)?;
777    if !is_slot(&schedule.rule, schedule.start_on, slot_on) {
778        bail!("error invalid-sync-change recurrence-slot-off-lattice");
779    }
780    let outcome = RecurrenceOutcome::parse(&required_string_payload("outcome", &change.payload)?)
781        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
782    let resolved_at = required_timestamp_payload("resolved_at", &change.payload)?;
783    if resolved_at < crate::recurrence::slot_values(&schedule, slot_on)?.boundary_at {
784        bail!("error invalid-sync-change recurrence-resolution-before-slot");
785    }
786    let conflict_resolution = change
787        .payload
788        .get("conflict_resolution")
789        .and_then(Value::as_bool)
790        .unwrap_or(false);
791    let task_id = required_string_payload("task_id", &change.payload)?;
792    ensure_sync_id("task_id", &task_id)?;
793    let workspace_id: WorkspaceId =
794        required_string_payload("workspace_id", &change.payload)?.parse()?;
795    let series_id: RecurrenceSeriesId = change.entity_id.parse().unwrap();
796    let identity = derive_occurrence_identity(&workspace_id, &series_id, &schedule, slot_on)?;
797    if task_id != identity.task_id.as_str() {
798        bail!("error invalid-sync-change recurrence-deterministic-mismatch field=task_id");
799    }
800    let expected_status = match outcome {
801        RecurrenceOutcome::Completed => "done",
802        RecurrenceOutcome::Skipped => "canceled",
803    };
804    if required_string_payload("task_status", &change.payload)? != expected_status {
805        bail!("error invalid-sync-change recurrence-outcome-status-mismatch");
806    }
807    let status_change_id = required_string_payload("task_status_change_id", &change.payload)?;
808    if !conflict_resolution || !status_change_id.is_empty() {
809        ensure_sync_id("task_status_change_id", &status_change_id)?;
810    }
811    let successor = required_string_payload("successor_task_id", &change.payload)?;
812    if !successor.is_empty() {
813        ensure_sync_id("successor_task_id", &successor)?;
814        let successor_slot = next_slot_after(&schedule.rule, schedule.start_on, slot_on)
815            .context("error invalid-sync-change recurrence-successor-out-of-range")?;
816        let successor_identity =
817            derive_occurrence_identity(&workspace_id, &series_id, &schedule, successor_slot)?;
818        if successor != successor_identity.task_id.as_str() {
819            bail!(
820                "error invalid-sync-change recurrence-deterministic-mismatch field=successor_task_id"
821            );
822        }
823    }
824    Ok(())
825}
826
827fn validate_recurrence_state(change: &ChangeWire) -> Result<()> {
828    validate_recurrence_entity(change)?;
829    if change.field.as_deref() != Some("state") {
830        bail!("error invalid-sync-change field=state");
831    }
832    let conflict_resolution = change
833        .payload
834        .get("conflict_resolution")
835        .and_then(Value::as_bool)
836        .unwrap_or(false);
837    if change.base_version.is_none() && !conflict_resolution {
838        bail!("error invalid-sync-change recurrence-lifecycle-base-version-missing");
839    }
840    let state = RecurrenceSeriesState::parse(&required_string_payload("state", &change.payload)?)
841        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
842    let stopped_at = required_string_payload("stopped_at", &change.payload)?;
843    if matches!(state, RecurrenceSeriesState::Stopped) {
844        required_timestamp_payload("stopped_at", &change.payload)?;
845    } else if !stopped_at.is_empty() {
846        bail!("error invalid-sync-change recurrence-stop-state-mismatch");
847    }
848    required_timestamp_payload("changed_at", &change.payload)?;
849    Ok(())
850}
851
852fn validate_recurrence_pause(change: &ChangeWire, open: bool) -> Result<()> {
853    validate_recurrence_entity(change)?;
854    if change.field.as_deref() != Some("pause") {
855        bail!("error invalid-sync-change field=pause");
856    }
857    if open {
858        let interval_id = required_string_payload("interval_id", &change.payload)?;
859        ensure_sync_id("interval_id", &interval_id)?;
860        required_timestamp_payload("paused_at", &change.payload)?;
861        let slot = required_string_payload("suspended_slot_on", &change.payload)?;
862        if !slot.is_empty() {
863            slot.parse::<NaiveDate>()
864                .context("error invalid-sync-change suspended_slot_on")?;
865            let task_id = required_string_payload("suspended_task_id", &change.payload)?;
866            ensure_sync_id("suspended_task_id", &task_id)?;
867        }
868    } else {
869        required_timestamp_payload("resumed_at", &change.payload)?;
870        if let Some(interval_id) = optional_string_payload("interval_id", &change.payload)? {
871            ensure_sync_id("interval_id", &interval_id)?;
872        }
873    }
874    Ok(())
875}
876
877fn validate_recurrence_entity(change: &ChangeWire) -> Result<()> {
878    ensure_entity_type(change, "recurrence_series")?;
879    change
880        .entity_id
881        .parse::<RecurrenceSeriesId>()
882        .map_err(|_| anyhow::anyhow!("error invalid-sync-change entity_id invalid-id"))?;
883    required_workspace_payload(&change.payload)?;
884    Ok(())
885}
886
887fn recurrence_schedule_payload(payload: &Value) -> Result<RecurrenceSchedule> {
888    let frequency = RecurrenceFrequency::parse(&required_string_payload("frequency", payload)?)
889        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
890    let interval = required_i64_payload("interval", payload)?;
891    let interval = u32::try_from(interval)
892        .context("error invalid-sync-change recurrence-interval-out-of-range")?;
893    let weekdays_text = required_string_payload("weekdays", payload)?;
894    let weekdays = weekdays_text
895        .parse::<WeekdaySet>()
896        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
897    if weekdays.to_string() != weekdays_text {
898        bail!("error invalid-sync-change recurrence-weekdays-noncanonical");
899    }
900    let rule = RecurrenceRule::new(frequency, interval, weekdays)
901        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
902    let timezone_text = required_string_payload("timezone", payload)?;
903    let timezone = timezone_text
904        .parse::<TimeZoneId>()
905        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
906    if timezone_text.parse::<chrono_tz::Tz>()?.to_string() != timezone_text {
907        bail!("error invalid-sync-change recurrence-timezone-noncanonical");
908    }
909    let start_on = recurrence_date_payload("start_on", payload)?;
910    let available = required_string_payload("available_local_time", payload)?;
911    let available_local_time = if available.is_empty() {
912        None
913    } else {
914        Some(validate_local_time(&available)?)
915    };
916    let due_policy = RecurrenceDuePolicy::parse(&required_string_payload("due_policy", payload)?)
917        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
918    Ok(RecurrenceSchedule::new(
919        rule,
920        timezone,
921        start_on,
922        available_local_time,
923        due_policy,
924    ))
925}
926
927fn recurrence_date_payload(key: &str, payload: &Value) -> Result<NaiveDate> {
928    required_string_payload(key, payload)?
929        .parse()
930        .with_context(|| format!("error invalid-sync-change payload.{key} invalid-date"))
931}
932
933fn validate_local_time(value: &str) -> Result<NaiveTime> {
934    NaiveTime::parse_from_str(value, "%H:%M:%S")
935        .context("error invalid-sync-change recurrence-local-time")
936}
937
938fn validate_attachment_add_change(change: &ChangeWire) -> Result<()> {
939    ensure_entity_type(change, "task")?;
940    ensure_sync_id("entity_id", &change.entity_id)?;
941    ensure_attachment_field(change)?;
942    required_workspace_payload(&change.payload)?;
943    let attachment_id = required_string_payload("attachment_id", &change.payload)?;
944    ensure_sync_id("attachment_id", &attachment_id)?;
945    let sha256 = required_string_payload("sha256", &change.payload)?;
946    validate_sha256_for_sync(&sha256)?;
947    let byte_size = required_i64_payload("byte_size", &change.payload)?;
948    validate_blob_size_for_sync(byte_size)?;
949    let media_type = required_string_payload("media_type", &change.payload)?;
950    map_attachment_validation(validate_media_type(&media_type))?;
951    let filename = optional_string_payload("filename", &change.payload)?;
952    map_attachment_validation(validate_filename(filename.as_deref()))?;
953    let alt_text = optional_string_payload("alt_text", &change.payload)?;
954    map_attachment_validation(validate_alt_text(alt_text.as_deref()))?;
955    let width = optional_i64_payload("width", &change.payload)?;
956    let height = optional_i64_payload("height", &change.payload)?;
957    map_attachment_validation(validate_dimensions(width, height))?;
958    required_timestamp_payload("created_at", &change.payload)?;
959    Ok(())
960}
961
962fn validate_attachment_delete_change(change: &ChangeWire) -> Result<()> {
963    ensure_entity_type(change, "task")?;
964    ensure_sync_id("entity_id", &change.entity_id)?;
965    ensure_attachment_field(change)?;
966    required_workspace_payload(&change.payload)?;
967    let attachment_id = required_string_payload("attachment_id", &change.payload)?;
968    ensure_sync_id("attachment_id", &attachment_id)?;
969    required_timestamp_payload("deleted_at", &change.payload)?;
970    Ok(())
971}
972
973fn ensure_attachment_field(change: &ChangeWire) -> Result<()> {
974    if change.field.as_deref() != Some("attachments") {
975        bail!("error invalid-sync-change field=attachments");
976    }
977    Ok(())
978}
979
980fn validate_sha256_for_sync(value: &str) -> Result<()> {
981    if value.len() == 64
982        && value
983            .bytes()
984            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
985    {
986        Ok(())
987    } else {
988        bail!("error invalid-sync-change invalid-sha256");
989    }
990}
991
992fn validate_blob_size_for_sync(byte_size: i64) -> Result<()> {
993    let Ok(bytes) = usize::try_from(byte_size) else {
994        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
995    };
996    if bytes == 0 || bytes > MAX_BLOB_BYTES {
997        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
998    }
999    map_attachment_validation(validate_blob_size(bytes))
1000}
1001
1002fn map_attachment_validation(result: Result<()>) -> Result<()> {
1003    result.map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
1004}
1005
1006fn validate_change_server_seq(change: &ChangeWire, direction: ChangeDirection) -> Result<()> {
1007    match direction {
1008        ChangeDirection::Pushed if change.server_seq.is_some() => {
1009            bail!("error invalid-sync-change server_seq client-supplied");
1010        }
1011        ChangeDirection::Pulled => match change.server_seq {
1012            Some(server_seq) if server_seq > 0 => {}
1013            Some(server_seq) => {
1014                bail!("error invalid-sync-change server_seq={server_seq}");
1015            }
1016            None => bail!("error invalid-sync-change server_seq missing"),
1017        },
1018        ChangeDirection::Pushed => {}
1019    }
1020    Ok(())
1021}
1022
1023fn validate_push_acks(request_change_ids: &[String], response: &SyncResponse) -> Result<()> {
1024    if response.push_acks.len() != request_change_ids.len() {
1025        bail!(
1026            "error invalid-sync-response push-ack-count expected={} got={}",
1027            request_change_ids.len(),
1028            response.push_acks.len()
1029        );
1030    }
1031    let expected = request_change_ids
1032        .iter()
1033        .map(String::as_str)
1034        .collect::<HashSet<_>>();
1035    let mut seen = HashSet::with_capacity(response.push_acks.len());
1036    for ack in &response.push_acks {
1037        if !expected.contains(ack.change_id.as_str()) {
1038            bail!(
1039                "error invalid-sync-response unexpected-push-ack change_id={}",
1040                ack.change_id
1041            );
1042        }
1043        if ack.server_seq <= 0 {
1044            bail!(
1045                "error invalid-sync-response push-ack-server-seq change_id={} server_seq={}",
1046                ack.change_id,
1047                ack.server_seq
1048            );
1049        }
1050        if !seen.insert(ack.change_id.as_str()) {
1051            bail!(
1052                "error invalid-sync-response duplicate-push-ack change_id={}",
1053                ack.change_id
1054            );
1055        }
1056    }
1057    Ok(())
1058}
1059
1060fn validate_pull_page(after: i64, pull_limit: u32, response: &SyncResponse) -> Result<()> {
1061    let mut previous = after;
1062    let mut change_ids = HashSet::with_capacity(response.changes.len());
1063    for change in &response.changes {
1064        validate_pulled_change(change)?;
1065        if !change_ids.insert(&change.change_id) {
1066            bail!(
1067                "error invalid-sync-response duplicate-pull-change change_id={}",
1068                change.change_id
1069            );
1070        }
1071        let server_seq = change.server_seq.with_context(|| {
1072            format!(
1073                "error invalid-sync-response missing-server-seq change_id={}",
1074                change.change_id
1075            )
1076        })?;
1077        if server_seq <= previous {
1078            bail!(
1079                "error invalid-sync-response server-seq-order previous={} server_seq={}",
1080                previous,
1081                server_seq
1082            );
1083        }
1084        previous = server_seq;
1085    }
1086    let expected_cursor = response
1087        .changes
1088        .last()
1089        .and_then(|change| change.server_seq)
1090        .unwrap_or(after);
1091    if response.cursor != expected_cursor {
1092        bail!(
1093            "error invalid-sync-response cursor-mismatch expected={} got={}",
1094            expected_cursor,
1095            response.cursor
1096        );
1097    }
1098    if response.has_more && response.changes.len() < pull_limit as usize {
1099        bail!(
1100            "error invalid-sync-response has-more-short-page returned={} limit={}",
1101            response.changes.len(),
1102            pull_limit
1103        );
1104    }
1105    Ok(())
1106}
1107
1108fn validate_push_pull_overlap(response: &SyncResponse) -> Result<()> {
1109    let acked = response
1110        .push_acks
1111        .iter()
1112        .map(|ack| (ack.change_id.as_str(), ack.server_seq))
1113        .collect::<HashMap<_, _>>();
1114    for change in &response.changes {
1115        if let Some(acked_server_seq) = acked.get(change.change_id.as_str()) {
1116            let Some(pull_server_seq) = change.server_seq else {
1117                continue;
1118            };
1119            if *acked_server_seq != pull_server_seq {
1120                bail!(
1121                    "error invalid-sync-response push-pull-server-seq-mismatch change_id={} ack={} pull={}",
1122                    change.change_id,
1123                    acked_server_seq,
1124                    pull_server_seq
1125                );
1126            }
1127        }
1128    }
1129    Ok(())
1130}
1131
1132fn validate_sync_task_field_value(field: TaskField, value: &str) -> Result<()> {
1133    field
1134        .validate_value(value)
1135        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
1136}
1137
1138fn ensure_entity_type(change: &ChangeWire, expected: &str) -> Result<()> {
1139    if change.entity_type == expected {
1140        Ok(())
1141    } else {
1142        bail!(
1143            "error invalid-sync-change op_type={} entity_type={} expected={}",
1144            change.op_type,
1145            change.entity_type,
1146            expected
1147        )
1148    }
1149}
1150
1151fn ensure_non_empty(name: &str, value: &str) -> Result<()> {
1152    if value.trim().is_empty() {
1153        bail!("error invalid-sync-change {name} empty");
1154    }
1155    Ok(())
1156}
1157
1158fn ensure_sync_id(name: &str, value: &str) -> Result<()> {
1159    if value.len() == 16 && value.bytes().all(|byte| BASE32.contains(&byte)) {
1160        Ok(())
1161    } else {
1162        bail!("error invalid-sync-change {name} invalid-id");
1163    }
1164}
1165
1166fn ensure_project_id(name: &str, value: &str) -> Result<()> {
1167    if value.parse::<ProjectId>().is_ok() {
1168        Ok(())
1169    } else {
1170        bail!("error invalid-sync-change {name} invalid-id");
1171    }
1172}
1173
1174fn required_string_payload(key: &str, payload: &Value) -> Result<String> {
1175    payload
1176        .get(key)
1177        .and_then(Value::as_str)
1178        .map(str::to_string)
1179        .with_context(|| format!("error invalid-sync-change payload.{key} missing"))
1180}
1181
1182fn required_i64_payload(key: &str, payload: &Value) -> Result<i64> {
1183    match payload.get(key) {
1184        Some(Value::Number(value)) => value
1185            .as_i64()
1186            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
1187        Some(Value::Null) | None => bail!("error invalid-sync-change payload.{key} missing"),
1188        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1189    }
1190}
1191
1192fn required_timestamp_payload(key: &str, payload: &Value) -> Result<String> {
1193    let value = required_string_payload(key, payload)?;
1194    validate_timestamp_value(&format!("payload.{key}"), &value)?;
1195    Ok(value)
1196}
1197
1198fn validate_timestamp_value(label: &str, value: &str) -> Result<()> {
1199    if value.ends_with('Z') && chrono::DateTime::parse_from_rfc3339(value).is_ok() {
1200        Ok(())
1201    } else {
1202        bail!("error invalid-sync-change {label} invalid-timestamp");
1203    }
1204}
1205
1206fn required_workspace_payload(payload: &Value) -> Result<()> {
1207    let id = required_string_payload("workspace_id", payload)?;
1208    if id.parse::<WorkspaceId>().is_err() {
1209        bail!("error invalid-sync-change workspace_id invalid-id");
1210    }
1211    required_string_payload("workspace_key", payload)?;
1212    Ok(())
1213}
1214
1215fn optional_workspace_payload(payload: &Value) -> Result<()> {
1216    if payload.get("workspace_id").is_none() && payload.get("workspace_key").is_none() {
1217        return Ok(());
1218    }
1219    required_workspace_payload(payload)
1220}
1221
1222fn optional_string_payload(key: &str, payload: &Value) -> Result<Option<String>> {
1223    match payload.get(key) {
1224        Some(Value::String(value)) => Ok(Some(value.clone())),
1225        Some(Value::Null) | None => Ok(None),
1226        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1227    }
1228}
1229
1230fn optional_i64_payload(key: &str, payload: &Value) -> Result<Option<i64>> {
1231    match payload.get(key) {
1232        Some(Value::Number(value)) => value
1233            .as_i64()
1234            .map(Some)
1235            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
1236        Some(Value::Null) | None => Ok(None),
1237        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1238    }
1239}
1240
1241fn string_array_payload(key: &str, payload: &Value) -> Result<Vec<String>> {
1242    payload
1243        .get(key)
1244        .and_then(Value::as_array)
1245        .with_context(|| format!("error invalid-sync-change payload.{key} invalid"))?
1246        .iter()
1247        .map(|value| {
1248            value
1249                .as_str()
1250                .map(str::to_string)
1251                .with_context(|| format!("error invalid-sync-change payload.{key} invalid"))
1252        })
1253        .collect()
1254}
1255
1256fn optional_string_array_payload(key: &str, payload: &Value) -> Result<()> {
1257    match payload.get(key) {
1258        Some(Value::Array(values))
1259            if values
1260                .iter()
1261                .all(|value| value.as_str().is_some_and(|value| !value.trim().is_empty())) =>
1262        {
1263            Ok(())
1264        }
1265        Some(Value::Null) | None => Ok(()),
1266        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1267    }
1268}
1269
1270#[cfg(test)]
1271mod tests {
1272    use super::*;
1273    use crate::change_log::{ChangePayload, op_type};
1274    use crate::workspaces::Workspace;
1275
1276    fn test_workspace() -> Workspace {
1277        Workspace {
1278            id: "0000000000000000".parse().unwrap(),
1279            key: "default".to_string(),
1280            name: "default".to_string(),
1281        }
1282    }
1283
1284    fn make_change_wire(
1285        op_type: &str,
1286        entity_type: &str,
1287        entity_id: &str,
1288        payload: serde_json::Value,
1289    ) -> ChangeWire {
1290        ChangeWire {
1291            change_id: "AAAAAAAAAAAAAAA0".to_string(),
1292            client_id: "client".to_string(),
1293            local_seq: 1,
1294            entity_type: entity_type.to_string(),
1295            entity_id: entity_id.to_string(),
1296            field: None,
1297            op_type: op_type.to_string(),
1298            payload,
1299            base_version: None,
1300            created_at: "2026-06-01T00:00:00Z".to_string(),
1301            server_seq: None,
1302        }
1303    }
1304
1305    #[test]
1306    fn sync_timestamps_accept_fractional_utc_precision() {
1307        validate_timestamp_value("created_at", "2026-07-28T06:41:44Z").unwrap();
1308        validate_timestamp_value("created_at", "2026-07-28T06:41:44.000000Z").unwrap();
1309        validate_timestamp_value("created_at", "2026-07-28T06:41:44.123456789Z").unwrap();
1310    }
1311
1312    #[test]
1313    fn sync_timestamps_reject_offsets_and_invalid_calendar_values() {
1314        for value in ["2026-07-28T08:41:44+02:00", "2026-13-28T06:41:44Z", "today"] {
1315            assert!(validate_timestamp_value("created_at", value).is_err());
1316        }
1317    }
1318
1319    #[test]
1320    fn recurrence_projection_rejects_nondeterministic_change_timestamp() {
1321        let workspace = test_workspace();
1322        let series_id: RecurrenceSeriesId = "AAAAAAAAAAAAAAAA".parse().unwrap();
1323        let schedule = RecurrenceSchedule::new(
1324            RecurrenceRule::daily(),
1325            "UTC".parse().unwrap(),
1326            "2026-07-20".parse().unwrap(),
1327            None,
1328            RecurrenceDuePolicy::SameDay,
1329        );
1330        let slot_on: NaiveDate = "2026-07-20".parse().unwrap();
1331        let identity =
1332            derive_occurrence_identity(&workspace.id, &series_id, &schedule, slot_on).unwrap();
1333        let payload = ChangePayload::workspace(&workspace)
1334            .set("series_id", series_id.as_str())
1335            .set("slot_on", slot_on.to_string())
1336            .set("task_id", identity.task_id.as_str())
1337            .set("projected_at", &identity.occurrence_link.projected_at)
1338            .set("task_change_id", &identity.task_change_id)
1339            .set("occurrence_change_id", &identity.occurrence_change_id)
1340            .set(
1341                "task_field_version_seed",
1342                &identity.field_version_seeds.task,
1343            )
1344            .set(
1345                "occurrence_field_version_seed",
1346                &identity.field_version_seeds.occurrence,
1347            )
1348            .set("frequency", "daily")
1349            .set("interval", 1)
1350            .set("weekdays", "")
1351            .set("timezone", "UTC")
1352            .set("start_on", "2026-07-20")
1353            .set("available_local_time", "")
1354            .set("due_policy", "same_day")
1355            .into_value();
1356        let mut change = make_change_wire(
1357            op_type::PROJECT_RECURRENCE_OCCURRENCE,
1358            "recurrence_series",
1359            series_id.as_str(),
1360            payload,
1361        );
1362        change.change_id = identity.occurrence_change_id;
1363        change.field = Some("projection".to_string());
1364        change.created_at = identity.occurrence_link.projected_at;
1365        validate_pushed_change(&change).unwrap();
1366
1367        change.created_at = "2026-07-20T00:00:01Z".to_string();
1368        assert!(
1369            validate_pushed_change(&change)
1370                .unwrap_err()
1371                .to_string()
1372                .contains("recurrence-deterministic-mismatch")
1373        );
1374    }
1375
1376    #[test]
1377    fn request_pull_limit_has_default_and_bounds() {
1378        assert_eq!(request_pull_limit(None).unwrap(), MAX_PULL_BATCH);
1379        assert!(request_pull_limit(Some(MAX_PULL_BATCH)).is_ok());
1380        assert_eq!(
1381            request_pull_limit(Some(0)).unwrap_err().to_string(),
1382            "error sync-pull-limit-out-of-range min=1 max=512 got=0"
1383        );
1384        assert_eq!(
1385            request_pull_limit(Some(MAX_PULL_BATCH + 1))
1386                .unwrap_err()
1387                .to_string(),
1388            "error sync-pull-limit-out-of-range min=1 max=512 got=513"
1389        );
1390    }
1391
1392    #[test]
1393    fn request_envelope_rejects_negative_cursor_and_oversized_push_batch() {
1394        let request = SyncRequest {
1395            protocol_version: Some(SYNC_PROTOCOL_VERSION),
1396            client_id: "test-client".to_string(),
1397            after: -1,
1398            pull_limit: Some(MAX_PULL_BATCH),
1399            changes: Vec::new(),
1400        };
1401        assert_eq!(
1402            validate_sync_request_envelope(&request)
1403                .unwrap_err()
1404                .to_string(),
1405            "error sync-after-out-of-range min=0 got=-1"
1406        );
1407
1408        let request = SyncRequest {
1409            protocol_version: Some(SYNC_PROTOCOL_VERSION),
1410            client_id: "test-client".to_string(),
1411            after: 0,
1412            pull_limit: Some(MAX_PULL_BATCH),
1413            changes: vec![
1414                ChangeWire {
1415                    change_id: "AAAAAAAAAAAAAAA0".to_string(),
1416                    client_id: "client".to_string(),
1417                    local_seq: 1,
1418                    entity_type: "task".to_string(),
1419                    entity_id: "BBBBBBBBBBBBBBBB".to_string(),
1420                    field: None,
1421                    op_type: "create_task".to_string(),
1422                    payload: serde_json::json!({"title":"oops","project_id":"0000000000000000","project_key":"app","project_name":"app","project_prefix":"APP","workspace_id":"0000000000000000","workspace_key":"default","created_at":"2026-01-01T00:00:00Z"}),
1423                    base_version: None,
1424                    created_at: "2026-01-01T00:00:00Z".to_string(),
1425                    server_seq: None,
1426                };
1427                MAX_PUSH_BATCH + 1
1428            ],
1429        };
1430        assert_eq!(
1431            validate_sync_request_envelope(&request)
1432                .unwrap_err()
1433                .to_string(),
1434            "error sync-push-too-large limit=256 got=257"
1435        );
1436    }
1437
1438    #[test]
1439    fn response_validation_respects_request_pull_limit() {
1440        let response = SyncResponse {
1441            protocol_version: SYNC_PROTOCOL_VERSION,
1442            cursor: 1,
1443            has_more: false,
1444            push_acks: vec![],
1445            changes: vec![
1446                ChangeWire {
1447                    change_id: "AAAAAAAAAAAAAAA1".to_string(),
1448                    client_id: "client".to_string(),
1449                    local_seq: 1,
1450                    entity_type: "task".to_string(),
1451                    entity_id: "BBBBBBBBBBBBBBBB".to_string(),
1452                    field: None,
1453                    op_type: "create_task".to_string(),
1454                    payload: serde_json::json!({
1455                        "title":"one",
1456                        "project_id":"0000000000000000",
1457                        "project_key":"app",
1458                        "project_name":"app",
1459                        "project_prefix":"APP",
1460                    }),
1461                    base_version: None,
1462                    created_at: "2026-01-01T00:00:00Z".to_string(),
1463                    server_seq: Some(1),
1464                };
1465                MAX_PULL_BATCH as usize + 1
1466            ],
1467        };
1468        assert_eq!(
1469            validate_sync_response_for_request(0, MAX_PULL_BATCH, &[], &response)
1470                .unwrap_err()
1471                .to_string(),
1472            "error invalid-sync-response pull-too-large limit=512 got=513"
1473        );
1474    }
1475
1476    #[test]
1477    fn constructed_create_task_payload_passes_wire_validation() {
1478        let ws = test_workspace();
1479        let payload = ChangePayload::workspace(&ws)
1480            .set("title", "test task")
1481            .set("description", "a description")
1482            .set("project_id", "1111111111111111")
1483            .set("project_key", "app")
1484            .set("project_name", "App")
1485            .set("project_prefix", "APP")
1486            .set("status", "inbox")
1487            .set("priority", "none")
1488            .set("created_at", "2026-06-01T00:00:00Z")
1489            .into_value();
1490        let change = make_change_wire(op_type::CREATE_TASK, "task", "BBBBBBBBBBBBBBBB", payload);
1491        validate_pushed_change(&change)
1492            .expect("create_task payload built with ChangePayload should be wire-valid");
1493    }
1494
1495    #[test]
1496    fn constructed_create_project_payload_passes_wire_validation() {
1497        let ws = test_workspace();
1498        let payload = ChangePayload::workspace(&ws)
1499            .set("key", "app")
1500            .set("name", "App")
1501            .set("prefix", "APP")
1502            .set("created_at", "2026-06-01T00:00:00Z")
1503            .into_value();
1504        let change = make_change_wire(
1505            op_type::CREATE_PROJECT,
1506            "project",
1507            "1111111111111111",
1508            payload,
1509        );
1510        validate_pushed_change(&change)
1511            .expect("create_project payload built with ChangePayload should be wire-valid");
1512    }
1513
1514    #[test]
1515    fn project_changes_reject_invalid_project_ids() {
1516        let ws = test_workspace();
1517        let project_payload = ChangePayload::workspace(&ws)
1518            .set("key", "app")
1519            .set("name", "App")
1520            .set("prefix", "APP")
1521            .set("created_at", "2026-06-01T00:00:00Z")
1522            .into_value();
1523        let create_project = make_change_wire(
1524            op_type::CREATE_PROJECT,
1525            "project",
1526            "invalid",
1527            project_payload,
1528        );
1529        assert_eq!(
1530            validate_pushed_change(&create_project)
1531                .unwrap_err()
1532                .to_string(),
1533            "error invalid-sync-change entity_id invalid-id"
1534        );
1535
1536        let task_payload = ChangePayload::workspace(&ws)
1537            .set("title", "test task")
1538            .set("project_id", "invalid")
1539            .set("project_key", "app")
1540            .set("project_name", "App")
1541            .set("project_prefix", "APP")
1542            .set("created_at", "2026-06-01T00:00:00Z")
1543            .into_value();
1544        let create_task = make_change_wire(
1545            op_type::CREATE_TASK,
1546            "task",
1547            "BBBBBBBBBBBBBBBB",
1548            task_payload,
1549        );
1550        assert_eq!(
1551            validate_pushed_change(&create_task)
1552                .unwrap_err()
1553                .to_string(),
1554            "error invalid-sync-change project_id invalid-id"
1555        );
1556
1557        let field_payload = ChangePayload::workspace(&ws)
1558            .set("value", "invalid")
1559            .set("project_id", "invalid")
1560            .set("project_key", "app")
1561            .set("project_name", "App")
1562            .set("project_prefix", "APP")
1563            .into_value();
1564        let mut set_field = make_change_wire(
1565            op_type::SET_FIELD,
1566            "task",
1567            "BBBBBBBBBBBBBBBB",
1568            field_payload,
1569        );
1570        set_field.field = Some("project".to_string());
1571        assert_eq!(
1572            validate_pushed_change(&set_field).unwrap_err().to_string(),
1573            "error invalid-sync-change project_id invalid-id"
1574        );
1575    }
1576
1577    #[test]
1578    fn constructed_label_add_payload_passes_wire_validation() {
1579        let ws = test_workspace();
1580        let payload = ChangePayload::workspace(&ws)
1581            .set("label", "bug")
1582            .into_value();
1583        let mut change = make_change_wire(op_type::LABEL_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1584        change.field = Some("labels".to_string());
1585        validate_pushed_change(&change)
1586            .expect("label_add payload built with ChangePayload should be wire-valid");
1587    }
1588
1589    #[test]
1590    fn constructed_label_administration_payloads_pass_wire_validation() {
1591        let ws = test_workspace();
1592        let rename = make_change_wire(
1593            op_type::SET_LABEL_NAME,
1594            "label",
1595            "old",
1596            ChangePayload::workspace(&ws)
1597                .set("name", "old")
1598                .set("new_name", "new")
1599                .set("renamed_at", "2026-06-01T00:00:00Z")
1600                .into_value(),
1601        );
1602        validate_pushed_change(&rename).unwrap();
1603
1604        let restore = make_change_wire(
1605            op_type::LABEL_RESTORE,
1606            "label",
1607            "new",
1608            ChangePayload::workspace(&ws)
1609                .set("name", "new")
1610                .set("created_at", "2026-06-01T00:00:00Z")
1611                .set("task_ids", ["BBBBBBBBBBBBBBBB"])
1612                .set("series_ids", ["CCCCCCCCCCCCCCCC"])
1613                .set("restored_at", "2026-06-01T00:00:01Z")
1614                .into_value(),
1615        );
1616        validate_pushed_change(&restore).unwrap();
1617    }
1618
1619    #[test]
1620    fn constructed_dependency_add_payload_passes_wire_validation() {
1621        let ws = test_workspace();
1622        let payload = ChangePayload::workspace(&ws)
1623            .set("depends_on_task_id", "CCCCCCCCCCCCCCCC")
1624            .into_value();
1625        let mut change =
1626            make_change_wire(op_type::DEPENDENCY_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1627        change.field = Some("dependencies".to_string());
1628        validate_pushed_change(&change)
1629            .expect("dependency_add payload built with ChangePayload should be wire-valid");
1630    }
1631
1632    #[test]
1633    fn constructed_note_edit_payload_passes_wire_validation() {
1634        let ws = test_workspace();
1635        let payload = ChangePayload::workspace(&ws)
1636            .set("note_id", "DDDDDDDDDDDDDDDD")
1637            .set("body", "corrected note body")
1638            .set("edited_at", "2026-06-01T01:00:00Z")
1639            .into_value();
1640        let mut change = make_change_wire(op_type::NOTE_EDIT, "task", "BBBBBBBBBBBBBBBB", payload);
1641        change.field = Some("notes".to_string());
1642        validate_pushed_change(&change)
1643            .expect("note_edit payload built with ChangePayload should be wire-valid");
1644    }
1645
1646    #[test]
1647    fn constructed_note_add_payload_passes_wire_validation() {
1648        let ws = test_workspace();
1649        let payload = ChangePayload::workspace(&ws)
1650            .set("note_id", "DDDDDDDDDDDDDDDD")
1651            .set("body", "note body")
1652            .set("created_at", "2026-06-01T00:00:00Z")
1653            .into_value();
1654        let mut change = make_change_wire(op_type::NOTE_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1655        change.field = Some("notes".to_string());
1656        validate_pushed_change(&change)
1657            .expect("note_add payload built with ChangePayload should be wire-valid");
1658    }
1659}
aven_core/sync/wire.rs

aven_core/sync/
wire.rs
