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 = 11;
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::LABEL_DELETE => {
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_timestamp_payload("deleted_at", &change.payload)?;
488        }
489        op_type::NOTE_DELETE => {
490            ensure_entity_type(change, "task")?;
491            ensure_sync_id("entity_id", &change.entity_id)?;
492            if change.field.as_deref() != Some("notes") {
493                bail!("error invalid-sync-change field=notes");
494            }
495            required_workspace_payload(&change.payload)?;
496            let note_id = required_string_payload("note_id", &change.payload)?;
497            ensure_sync_id("note_id", &note_id)?;
498            required_timestamp_payload("deleted_at", &change.payload)?;
499        }
500        op_type::CREATE_RECURRENCE_SERIES => validate_recurrence_create(change)?,
501        op_type::UPDATE_RECURRENCE_TEMPLATE => validate_recurrence_template(change)?,
502        op_type::PROJECT_RECURRENCE_OCCURRENCE => validate_recurrence_projection(change)?,
503        op_type::RESOLVE_RECURRENCE_OCCURRENCE => validate_recurrence_outcome(change)?,
504        op_type::SET_RECURRENCE_STATE | op_type::STOP_RECURRENCE_SERIES => {
505            validate_recurrence_state(change)?
506        }
507        op_type::OPEN_RECURRENCE_PAUSE => validate_recurrence_pause(change, true)?,
508        op_type::CLOSE_RECURRENCE_PAUSE => validate_recurrence_pause(change, false)?,
509        op_type::ATTACHMENT_ADD => validate_attachment_add_change(change)?,
510        op_type::ATTACHMENT_DELETE => validate_attachment_delete_change(change)?,
511        _ => bail!("error invalid-sync-change op_type={}", change.op_type),
512    }
513    Ok(())
514}
515
516fn validate_recurrence_task(change: &ChangeWire) -> Result<()> {
517    let workspace_id: WorkspaceId = required_string_payload("workspace_id", &change.payload)?
518        .parse()
519        .map_err(|_| anyhow::anyhow!("error invalid-sync-change workspace_id invalid-id"))?;
520    let series_id: RecurrenceSeriesId = required_string_payload("series_id", &change.payload)?
521        .parse()
522        .map_err(|_| anyhow::anyhow!("error invalid-sync-change series_id invalid-id"))?;
523    let slot_on = recurrence_date_payload("slot_on", &change.payload)?;
524    let schedule = recurrence_schedule_payload(&change.payload)?;
525    if !is_slot(&schedule.rule, schedule.start_on, slot_on) {
526        bail!("error invalid-sync-change recurrence-slot-off-lattice");
527    }
528    let identity = derive_occurrence_identity(&workspace_id, &series_id, &schedule, slot_on)
529        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
530    let slot = crate::recurrence::slot_values(&schedule, slot_on)
531        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
532    for (key, expected) in [
533        ("task_id", identity.task_id.as_str()),
534        ("created_at", identity.created_at.as_str()),
535        ("updated_at", identity.updated_at.as_str()),
536        ("available_at", slot.available_at.as_str()),
537        ("due_on", slot.due_on.as_deref().unwrap_or("")),
538        ("task_change_id", identity.task_change_id.as_str()),
539        (
540            "occurrence_change_id",
541            identity.occurrence_change_id.as_str(),
542        ),
543        (
544            "task_field_version_seed",
545            identity.field_version_seeds.task.as_str(),
546        ),
547        (
548            "occurrence_field_version_seed",
549            identity.field_version_seeds.occurrence.as_str(),
550        ),
551    ] {
552        if required_string_payload(key, &change.payload)? != expected {
553            bail!("error invalid-sync-change recurrence-deterministic-mismatch field={key}");
554        }
555    }
556    if change.entity_id != identity.task_id.as_str()
557        || change.change_id != identity.task_change_id
558        || change.created_at != identity.created_at
559    {
560        bail!("error invalid-sync-change recurrence-deterministic-mismatch field=change_identity");
561    }
562    Ok(())
563}
564
565fn validate_recurrence_create(change: &ChangeWire) -> Result<()> {
566    ensure_entity_type(change, "recurrence_series")?;
567    let series_id: RecurrenceSeriesId = change
568        .entity_id
569        .parse()
570        .map_err(|_| anyhow::anyhow!("error invalid-sync-change entity_id invalid-id"))?;
571    required_workspace_payload(&change.payload)?;
572    if required_string_payload("series_id", &change.payload)? != series_id.as_str() {
573        bail!("error invalid-sync-change recurrence-series-id-mismatch");
574    }
575    required_string_payload("title", &change.payload)?;
576    required_string_payload("description", &change.payload)?;
577    let project_id = required_string_payload("project_id", &change.payload)?;
578    ensure_project_id("project_id", &project_id)?;
579    required_string_payload("project_key", &change.payload)?;
580    required_string_payload("project_name", &change.payload)?;
581    required_string_payload("project_prefix", &change.payload)?;
582    validate_sync_task_field_value(
583        TaskField::Priority,
584        &required_string_payload("priority", &change.payload)?,
585    )?;
586    let status = required_string_payload("initial_status", &change.payload)?;
587    validate_sync_task_field_value(TaskField::Status, &status)?;
588    if matches!(status.as_str(), "done" | "canceled") {
589        bail!("error invalid-sync-change recurrence-terminal-template");
590    }
591    recurrence_schedule_payload(&change.payload)?;
592    optional_string_array_payload("labels", &change.payload)?;
593    if required_string_payload("state", &change.payload)? != "active"
594        || !required_string_payload("stopped_at", &change.payload)?.is_empty()
595    {
596        bail!("error invalid-sync-change recurrence-create-state");
597    }
598    required_timestamp_payload("created_at", &change.payload)?;
599    required_timestamp_payload("updated_at", &change.payload)?;
600    Ok(())
601}
602
603fn validate_recurrence_template(change: &ChangeWire) -> Result<()> {
604    validate_recurrence_entity(change)?;
605    let fields = change
606        .payload
607        .get("fields")
608        .and_then(Value::as_array)
609        .context("error invalid-sync-change payload.fields missing")?;
610    if fields.len() > 8 {
611        bail!("error invalid-sync-change recurrence-template-fields-too-large");
612    }
613    let base_versions = change
614        .payload
615        .get("base_versions")
616        .and_then(Value::as_object)
617        .context("error invalid-sync-change payload.base_versions missing")?;
618    let mut seen = HashSet::new();
619    for pair in fields {
620        let pair = pair
621            .as_array()
622            .filter(|pair| pair.len() == 2)
623            .context("error invalid-sync-change recurrence-template-field-pair")?;
624        let field = pair[0]
625            .as_str()
626            .context("error invalid-sync-change recurrence-template-field")?;
627        let value = pair[1]
628            .as_str()
629            .context("error invalid-sync-change recurrence-template-value")?;
630        if !seen.insert(field) {
631            bail!("error invalid-sync-change recurrence-template-duplicate-field");
632        }
633        if !matches!(
634            base_versions.get(field),
635            Some(Value::String(_)) | Some(Value::Null)
636        ) {
637            bail!("error invalid-sync-change recurrence-template-base-version field={field}");
638        }
639        match field {
640            "title" | "description" => {}
641            "project" => ensure_project_id("project_id", value)?,
642            "priority" => validate_sync_task_field_value(TaskField::Priority, value)?,
643            "initial_status" => {
644                validate_sync_task_field_value(TaskField::Status, value)?;
645                if matches!(value, "done" | "canceled") {
646                    bail!("error invalid-sync-change recurrence-terminal-template");
647                }
648            }
649            "available_local_time" => {
650                validate_local_time(value)?;
651            }
652            "due_policy" => {
653                RecurrenceDuePolicy::parse(value)
654                    .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
655            }
656            _ => bail!("error invalid-sync-change recurrence-template-field={field}"),
657        }
658    }
659    optional_string_array_payload("labels", &change.payload)?;
660    let labels_changed = change
661        .payload
662        .get("labels_changed")
663        .and_then(Value::as_bool)
664        .context("error invalid-sync-change payload.labels_changed missing")?;
665    if labels_changed
666        && !matches!(
667            base_versions.get("labels"),
668            Some(Value::String(_)) | Some(Value::Null)
669        )
670    {
671        bail!("error invalid-sync-change recurrence-template-base-version field=labels");
672    }
673    required_timestamp_payload("updated_at", &change.payload)?;
674    Ok(())
675}
676
677fn validate_recurrence_projection(change: &ChangeWire) -> Result<()> {
678    validate_recurrence_entity(change)?;
679    if change.field.as_deref() != Some("projection") {
680        bail!("error invalid-sync-change field=projection");
681    }
682    let workspace_id: WorkspaceId = required_string_payload("workspace_id", &change.payload)?
683        .parse()
684        .map_err(|_| anyhow::anyhow!("error invalid-sync-change workspace_id invalid-id"))?;
685    let series_id: RecurrenceSeriesId = change.entity_id.parse().unwrap();
686    let slot_on = recurrence_date_payload("slot_on", &change.payload)?;
687    let schedule = recurrence_schedule_payload(&change.payload)?;
688    if !is_slot(&schedule.rule, schedule.start_on, slot_on) {
689        bail!("error invalid-sync-change recurrence-slot-off-lattice");
690    }
691    let identity = derive_occurrence_identity(&workspace_id, &series_id, &schedule, slot_on)
692        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
693    for (key, expected) in [
694        ("task_id", identity.task_id.as_str()),
695        (
696            "projected_at",
697            identity.occurrence_link.projected_at.as_str(),
698        ),
699        ("task_change_id", identity.task_change_id.as_str()),
700        (
701            "occurrence_change_id",
702            identity.occurrence_change_id.as_str(),
703        ),
704        (
705            "task_field_version_seed",
706            identity.field_version_seeds.task.as_str(),
707        ),
708        (
709            "occurrence_field_version_seed",
710            identity.field_version_seeds.occurrence.as_str(),
711        ),
712    ] {
713        if required_string_payload(key, &change.payload)? != expected {
714            bail!("error invalid-sync-change recurrence-deterministic-mismatch field={key}");
715        }
716    }
717    if change.change_id != identity.occurrence_change_id
718        || change.created_at != identity.occurrence_link.projected_at
719    {
720        bail!("error invalid-sync-change recurrence-deterministic-mismatch field=change_id");
721    }
722    Ok(())
723}
724
725fn validate_recurrence_outcome(change: &ChangeWire) -> Result<()> {
726    validate_recurrence_entity(change)?;
727    validate_timestamp_value("created_at", &change.created_at)?;
728    if change.field.as_deref() != Some("outcome") {
729        bail!("error invalid-sync-change field=outcome");
730    }
731    let slot_on = recurrence_date_payload("slot_on", &change.payload)?;
732    let schedule = recurrence_schedule_payload(&change.payload)?;
733    if !is_slot(&schedule.rule, schedule.start_on, slot_on) {
734        bail!("error invalid-sync-change recurrence-slot-off-lattice");
735    }
736    let outcome = RecurrenceOutcome::parse(&required_string_payload("outcome", &change.payload)?)
737        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
738    let resolved_at = required_timestamp_payload("resolved_at", &change.payload)?;
739    if resolved_at < crate::recurrence::slot_values(&schedule, slot_on)?.boundary_at {
740        bail!("error invalid-sync-change recurrence-resolution-before-slot");
741    }
742    let conflict_resolution = change
743        .payload
744        .get("conflict_resolution")
745        .and_then(Value::as_bool)
746        .unwrap_or(false);
747    let task_id = required_string_payload("task_id", &change.payload)?;
748    ensure_sync_id("task_id", &task_id)?;
749    let workspace_id: WorkspaceId =
750        required_string_payload("workspace_id", &change.payload)?.parse()?;
751    let series_id: RecurrenceSeriesId = change.entity_id.parse().unwrap();
752    let identity = derive_occurrence_identity(&workspace_id, &series_id, &schedule, slot_on)?;
753    if task_id != identity.task_id.as_str() {
754        bail!("error invalid-sync-change recurrence-deterministic-mismatch field=task_id");
755    }
756    let expected_status = match outcome {
757        RecurrenceOutcome::Completed => "done",
758        RecurrenceOutcome::Skipped => "canceled",
759    };
760    if required_string_payload("task_status", &change.payload)? != expected_status {
761        bail!("error invalid-sync-change recurrence-outcome-status-mismatch");
762    }
763    let status_change_id = required_string_payload("task_status_change_id", &change.payload)?;
764    if !conflict_resolution || !status_change_id.is_empty() {
765        ensure_sync_id("task_status_change_id", &status_change_id)?;
766    }
767    let successor = required_string_payload("successor_task_id", &change.payload)?;
768    if !successor.is_empty() {
769        ensure_sync_id("successor_task_id", &successor)?;
770        let successor_slot = next_slot_after(&schedule.rule, schedule.start_on, slot_on)
771            .context("error invalid-sync-change recurrence-successor-out-of-range")?;
772        let successor_identity =
773            derive_occurrence_identity(&workspace_id, &series_id, &schedule, successor_slot)?;
774        if successor != successor_identity.task_id.as_str() {
775            bail!(
776                "error invalid-sync-change recurrence-deterministic-mismatch field=successor_task_id"
777            );
778        }
779    }
780    Ok(())
781}
782
783fn validate_recurrence_state(change: &ChangeWire) -> Result<()> {
784    validate_recurrence_entity(change)?;
785    if change.field.as_deref() != Some("state") {
786        bail!("error invalid-sync-change field=state");
787    }
788    let conflict_resolution = change
789        .payload
790        .get("conflict_resolution")
791        .and_then(Value::as_bool)
792        .unwrap_or(false);
793    if change.base_version.is_none() && !conflict_resolution {
794        bail!("error invalid-sync-change recurrence-lifecycle-base-version-missing");
795    }
796    let state = RecurrenceSeriesState::parse(&required_string_payload("state", &change.payload)?)
797        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
798    let stopped_at = required_string_payload("stopped_at", &change.payload)?;
799    if matches!(state, RecurrenceSeriesState::Stopped) {
800        required_timestamp_payload("stopped_at", &change.payload)?;
801    } else if !stopped_at.is_empty() {
802        bail!("error invalid-sync-change recurrence-stop-state-mismatch");
803    }
804    required_timestamp_payload("changed_at", &change.payload)?;
805    Ok(())
806}
807
808fn validate_recurrence_pause(change: &ChangeWire, open: bool) -> Result<()> {
809    validate_recurrence_entity(change)?;
810    if change.field.as_deref() != Some("pause") {
811        bail!("error invalid-sync-change field=pause");
812    }
813    if open {
814        let interval_id = required_string_payload("interval_id", &change.payload)?;
815        ensure_sync_id("interval_id", &interval_id)?;
816        required_timestamp_payload("paused_at", &change.payload)?;
817        let slot = required_string_payload("suspended_slot_on", &change.payload)?;
818        if !slot.is_empty() {
819            slot.parse::<NaiveDate>()
820                .context("error invalid-sync-change suspended_slot_on")?;
821            let task_id = required_string_payload("suspended_task_id", &change.payload)?;
822            ensure_sync_id("suspended_task_id", &task_id)?;
823        }
824    } else {
825        required_timestamp_payload("resumed_at", &change.payload)?;
826        if let Some(interval_id) = optional_string_payload("interval_id", &change.payload)? {
827            ensure_sync_id("interval_id", &interval_id)?;
828        }
829    }
830    Ok(())
831}
832
833fn validate_recurrence_entity(change: &ChangeWire) -> Result<()> {
834    ensure_entity_type(change, "recurrence_series")?;
835    change
836        .entity_id
837        .parse::<RecurrenceSeriesId>()
838        .map_err(|_| anyhow::anyhow!("error invalid-sync-change entity_id invalid-id"))?;
839    required_workspace_payload(&change.payload)?;
840    Ok(())
841}
842
843fn recurrence_schedule_payload(payload: &Value) -> Result<RecurrenceSchedule> {
844    let frequency = RecurrenceFrequency::parse(&required_string_payload("frequency", payload)?)
845        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
846    let interval = required_i64_payload("interval", payload)?;
847    let interval = u32::try_from(interval)
848        .context("error invalid-sync-change recurrence-interval-out-of-range")?;
849    let weekdays_text = required_string_payload("weekdays", payload)?;
850    let weekdays = weekdays_text
851        .parse::<WeekdaySet>()
852        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
853    if weekdays.to_string() != weekdays_text {
854        bail!("error invalid-sync-change recurrence-weekdays-noncanonical");
855    }
856    let rule = RecurrenceRule::new(frequency, interval, weekdays)
857        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
858    let timezone_text = required_string_payload("timezone", payload)?;
859    let timezone = timezone_text
860        .parse::<TimeZoneId>()
861        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
862    if timezone_text.parse::<chrono_tz::Tz>()?.to_string() != timezone_text {
863        bail!("error invalid-sync-change recurrence-timezone-noncanonical");
864    }
865    let start_on = recurrence_date_payload("start_on", payload)?;
866    let available = required_string_payload("available_local_time", payload)?;
867    let available_local_time = if available.is_empty() {
868        None
869    } else {
870        Some(validate_local_time(&available)?)
871    };
872    let due_policy = RecurrenceDuePolicy::parse(&required_string_payload("due_policy", payload)?)
873        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))?;
874    Ok(RecurrenceSchedule::new(
875        rule,
876        timezone,
877        start_on,
878        available_local_time,
879        due_policy,
880    ))
881}
882
883fn recurrence_date_payload(key: &str, payload: &Value) -> Result<NaiveDate> {
884    required_string_payload(key, payload)?
885        .parse()
886        .with_context(|| format!("error invalid-sync-change payload.{key} invalid-date"))
887}
888
889fn validate_local_time(value: &str) -> Result<NaiveTime> {
890    NaiveTime::parse_from_str(value, "%H:%M:%S")
891        .context("error invalid-sync-change recurrence-local-time")
892}
893
894fn validate_attachment_add_change(change: &ChangeWire) -> Result<()> {
895    ensure_entity_type(change, "task")?;
896    ensure_sync_id("entity_id", &change.entity_id)?;
897    ensure_attachment_field(change)?;
898    required_workspace_payload(&change.payload)?;
899    let attachment_id = required_string_payload("attachment_id", &change.payload)?;
900    ensure_sync_id("attachment_id", &attachment_id)?;
901    let sha256 = required_string_payload("sha256", &change.payload)?;
902    validate_sha256_for_sync(&sha256)?;
903    let byte_size = required_i64_payload("byte_size", &change.payload)?;
904    validate_blob_size_for_sync(byte_size)?;
905    let media_type = required_string_payload("media_type", &change.payload)?;
906    map_attachment_validation(validate_media_type(&media_type))?;
907    let filename = optional_string_payload("filename", &change.payload)?;
908    map_attachment_validation(validate_filename(filename.as_deref()))?;
909    let alt_text = optional_string_payload("alt_text", &change.payload)?;
910    map_attachment_validation(validate_alt_text(alt_text.as_deref()))?;
911    let width = optional_i64_payload("width", &change.payload)?;
912    let height = optional_i64_payload("height", &change.payload)?;
913    map_attachment_validation(validate_dimensions(width, height))?;
914    required_timestamp_payload("created_at", &change.payload)?;
915    Ok(())
916}
917
918fn validate_attachment_delete_change(change: &ChangeWire) -> Result<()> {
919    ensure_entity_type(change, "task")?;
920    ensure_sync_id("entity_id", &change.entity_id)?;
921    ensure_attachment_field(change)?;
922    required_workspace_payload(&change.payload)?;
923    let attachment_id = required_string_payload("attachment_id", &change.payload)?;
924    ensure_sync_id("attachment_id", &attachment_id)?;
925    required_timestamp_payload("deleted_at", &change.payload)?;
926    Ok(())
927}
928
929fn ensure_attachment_field(change: &ChangeWire) -> Result<()> {
930    if change.field.as_deref() != Some("attachments") {
931        bail!("error invalid-sync-change field=attachments");
932    }
933    Ok(())
934}
935
936fn validate_sha256_for_sync(value: &str) -> Result<()> {
937    if value.len() == 64
938        && value
939            .bytes()
940            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
941    {
942        Ok(())
943    } else {
944        bail!("error invalid-sync-change invalid-sha256");
945    }
946}
947
948fn validate_blob_size_for_sync(byte_size: i64) -> Result<()> {
949    let Ok(bytes) = usize::try_from(byte_size) else {
950        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
951    };
952    if bytes == 0 || bytes > MAX_BLOB_BYTES {
953        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
954    }
955    map_attachment_validation(validate_blob_size(bytes))
956}
957
958fn map_attachment_validation(result: Result<()>) -> Result<()> {
959    result.map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
960}
961
962fn validate_change_server_seq(change: &ChangeWire, direction: ChangeDirection) -> Result<()> {
963    match direction {
964        ChangeDirection::Pushed if change.server_seq.is_some() => {
965            bail!("error invalid-sync-change server_seq client-supplied");
966        }
967        ChangeDirection::Pulled => match change.server_seq {
968            Some(server_seq) if server_seq > 0 => {}
969            Some(server_seq) => {
970                bail!("error invalid-sync-change server_seq={server_seq}");
971            }
972            None => bail!("error invalid-sync-change server_seq missing"),
973        },
974        ChangeDirection::Pushed => {}
975    }
976    Ok(())
977}
978
979fn validate_push_acks(request_change_ids: &[String], response: &SyncResponse) -> Result<()> {
980    if response.push_acks.len() != request_change_ids.len() {
981        bail!(
982            "error invalid-sync-response push-ack-count expected={} got={}",
983            request_change_ids.len(),
984            response.push_acks.len()
985        );
986    }
987    let expected = request_change_ids
988        .iter()
989        .map(String::as_str)
990        .collect::<HashSet<_>>();
991    let mut seen = HashSet::with_capacity(response.push_acks.len());
992    for ack in &response.push_acks {
993        if !expected.contains(ack.change_id.as_str()) {
994            bail!(
995                "error invalid-sync-response unexpected-push-ack change_id={}",
996                ack.change_id
997            );
998        }
999        if ack.server_seq <= 0 {
1000            bail!(
1001                "error invalid-sync-response push-ack-server-seq change_id={} server_seq={}",
1002                ack.change_id,
1003                ack.server_seq
1004            );
1005        }
1006        if !seen.insert(ack.change_id.as_str()) {
1007            bail!(
1008                "error invalid-sync-response duplicate-push-ack change_id={}",
1009                ack.change_id
1010            );
1011        }
1012    }
1013    Ok(())
1014}
1015
1016fn validate_pull_page(after: i64, pull_limit: u32, response: &SyncResponse) -> Result<()> {
1017    let mut previous = after;
1018    let mut change_ids = HashSet::with_capacity(response.changes.len());
1019    for change in &response.changes {
1020        validate_pulled_change(change)?;
1021        if !change_ids.insert(&change.change_id) {
1022            bail!(
1023                "error invalid-sync-response duplicate-pull-change change_id={}",
1024                change.change_id
1025            );
1026        }
1027        let server_seq = change.server_seq.with_context(|| {
1028            format!(
1029                "error invalid-sync-response missing-server-seq change_id={}",
1030                change.change_id
1031            )
1032        })?;
1033        if server_seq <= previous {
1034            bail!(
1035                "error invalid-sync-response server-seq-order previous={} server_seq={}",
1036                previous,
1037                server_seq
1038            );
1039        }
1040        previous = server_seq;
1041    }
1042    let expected_cursor = response
1043        .changes
1044        .last()
1045        .and_then(|change| change.server_seq)
1046        .unwrap_or(after);
1047    if response.cursor != expected_cursor {
1048        bail!(
1049            "error invalid-sync-response cursor-mismatch expected={} got={}",
1050            expected_cursor,
1051            response.cursor
1052        );
1053    }
1054    if response.has_more && response.changes.len() < pull_limit as usize {
1055        bail!(
1056            "error invalid-sync-response has-more-short-page returned={} limit={}",
1057            response.changes.len(),
1058            pull_limit
1059        );
1060    }
1061    Ok(())
1062}
1063
1064fn validate_push_pull_overlap(response: &SyncResponse) -> Result<()> {
1065    let acked = response
1066        .push_acks
1067        .iter()
1068        .map(|ack| (ack.change_id.as_str(), ack.server_seq))
1069        .collect::<HashMap<_, _>>();
1070    for change in &response.changes {
1071        if let Some(acked_server_seq) = acked.get(change.change_id.as_str()) {
1072            let Some(pull_server_seq) = change.server_seq else {
1073                continue;
1074            };
1075            if *acked_server_seq != pull_server_seq {
1076                bail!(
1077                    "error invalid-sync-response push-pull-server-seq-mismatch change_id={} ack={} pull={}",
1078                    change.change_id,
1079                    acked_server_seq,
1080                    pull_server_seq
1081                );
1082            }
1083        }
1084    }
1085    Ok(())
1086}
1087
1088fn validate_sync_task_field_value(field: TaskField, value: &str) -> Result<()> {
1089    field
1090        .validate_value(value)
1091        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
1092}
1093
1094fn ensure_entity_type(change: &ChangeWire, expected: &str) -> Result<()> {
1095    if change.entity_type == expected {
1096        Ok(())
1097    } else {
1098        bail!(
1099            "error invalid-sync-change op_type={} entity_type={} expected={}",
1100            change.op_type,
1101            change.entity_type,
1102            expected
1103        )
1104    }
1105}
1106
1107fn ensure_non_empty(name: &str, value: &str) -> Result<()> {
1108    if value.trim().is_empty() {
1109        bail!("error invalid-sync-change {name} empty");
1110    }
1111    Ok(())
1112}
1113
1114fn ensure_sync_id(name: &str, value: &str) -> Result<()> {
1115    if value.len() == 16 && value.bytes().all(|byte| BASE32.contains(&byte)) {
1116        Ok(())
1117    } else {
1118        bail!("error invalid-sync-change {name} invalid-id");
1119    }
1120}
1121
1122fn ensure_project_id(name: &str, value: &str) -> Result<()> {
1123    if value.parse::<ProjectId>().is_ok() {
1124        Ok(())
1125    } else {
1126        bail!("error invalid-sync-change {name} invalid-id");
1127    }
1128}
1129
1130fn required_string_payload(key: &str, payload: &Value) -> Result<String> {
1131    payload
1132        .get(key)
1133        .and_then(Value::as_str)
1134        .map(str::to_string)
1135        .with_context(|| format!("error invalid-sync-change payload.{key} missing"))
1136}
1137
1138fn required_i64_payload(key: &str, payload: &Value) -> Result<i64> {
1139    match payload.get(key) {
1140        Some(Value::Number(value)) => value
1141            .as_i64()
1142            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
1143        Some(Value::Null) | None => bail!("error invalid-sync-change payload.{key} missing"),
1144        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1145    }
1146}
1147
1148fn required_timestamp_payload(key: &str, payload: &Value) -> Result<String> {
1149    let value = required_string_payload(key, payload)?;
1150    validate_timestamp_value(&format!("payload.{key}"), &value)?;
1151    Ok(value)
1152}
1153
1154fn validate_timestamp_value(label: &str, value: &str) -> Result<()> {
1155    if value.ends_with('Z') && chrono::DateTime::parse_from_rfc3339(value).is_ok() {
1156        Ok(())
1157    } else {
1158        bail!("error invalid-sync-change {label} invalid-timestamp");
1159    }
1160}
1161
1162fn required_workspace_payload(payload: &Value) -> Result<()> {
1163    let id = required_string_payload("workspace_id", payload)?;
1164    if id.parse::<WorkspaceId>().is_err() {
1165        bail!("error invalid-sync-change workspace_id invalid-id");
1166    }
1167    required_string_payload("workspace_key", payload)?;
1168    Ok(())
1169}
1170
1171fn optional_workspace_payload(payload: &Value) -> Result<()> {
1172    if payload.get("workspace_id").is_none() && payload.get("workspace_key").is_none() {
1173        return Ok(());
1174    }
1175    required_workspace_payload(payload)
1176}
1177
1178fn optional_string_payload(key: &str, payload: &Value) -> Result<Option<String>> {
1179    match payload.get(key) {
1180        Some(Value::String(value)) => Ok(Some(value.clone())),
1181        Some(Value::Null) | None => Ok(None),
1182        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1183    }
1184}
1185
1186fn optional_i64_payload(key: &str, payload: &Value) -> Result<Option<i64>> {
1187    match payload.get(key) {
1188        Some(Value::Number(value)) => value
1189            .as_i64()
1190            .map(Some)
1191            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
1192        Some(Value::Null) | None => Ok(None),
1193        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1194    }
1195}
1196
1197fn optional_string_array_payload(key: &str, payload: &Value) -> Result<()> {
1198    match payload.get(key) {
1199        Some(Value::Array(values))
1200            if values
1201                .iter()
1202                .all(|value| value.as_str().is_some_and(|value| !value.trim().is_empty())) =>
1203        {
1204            Ok(())
1205        }
1206        Some(Value::Null) | None => Ok(()),
1207        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
1208    }
1209}
1210
1211#[cfg(test)]
1212mod tests {
1213    use super::*;
1214    use crate::change_log::{ChangePayload, op_type};
1215    use crate::workspaces::Workspace;
1216
1217    fn test_workspace() -> Workspace {
1218        Workspace {
1219            id: "0000000000000000".parse().unwrap(),
1220            key: "default".to_string(),
1221            name: "default".to_string(),
1222        }
1223    }
1224
1225    fn make_change_wire(
1226        op_type: &str,
1227        entity_type: &str,
1228        entity_id: &str,
1229        payload: serde_json::Value,
1230    ) -> ChangeWire {
1231        ChangeWire {
1232            change_id: "AAAAAAAAAAAAAAA0".to_string(),
1233            client_id: "client".to_string(),
1234            local_seq: 1,
1235            entity_type: entity_type.to_string(),
1236            entity_id: entity_id.to_string(),
1237            field: None,
1238            op_type: op_type.to_string(),
1239            payload,
1240            base_version: None,
1241            created_at: "2026-06-01T00:00:00Z".to_string(),
1242            server_seq: None,
1243        }
1244    }
1245
1246    #[test]
1247    fn sync_timestamps_accept_fractional_utc_precision() {
1248        validate_timestamp_value("created_at", "2026-07-28T06:41:44Z").unwrap();
1249        validate_timestamp_value("created_at", "2026-07-28T06:41:44.000000Z").unwrap();
1250        validate_timestamp_value("created_at", "2026-07-28T06:41:44.123456789Z").unwrap();
1251    }
1252
1253    #[test]
1254    fn sync_timestamps_reject_offsets_and_invalid_calendar_values() {
1255        for value in ["2026-07-28T08:41:44+02:00", "2026-13-28T06:41:44Z", "today"] {
1256            assert!(validate_timestamp_value("created_at", value).is_err());
1257        }
1258    }
1259
1260    #[test]
1261    fn recurrence_projection_rejects_nondeterministic_change_timestamp() {
1262        let workspace = test_workspace();
1263        let series_id: RecurrenceSeriesId = "AAAAAAAAAAAAAAAA".parse().unwrap();
1264        let schedule = RecurrenceSchedule::new(
1265            RecurrenceRule::daily(),
1266            "UTC".parse().unwrap(),
1267            "2026-07-20".parse().unwrap(),
1268            None,
1269            RecurrenceDuePolicy::SameDay,
1270        );
1271        let slot_on: NaiveDate = "2026-07-20".parse().unwrap();
1272        let identity =
1273            derive_occurrence_identity(&workspace.id, &series_id, &schedule, slot_on).unwrap();
1274        let payload = ChangePayload::workspace(&workspace)
1275            .set("series_id", series_id.as_str())
1276            .set("slot_on", slot_on.to_string())
1277            .set("task_id", identity.task_id.as_str())
1278            .set("projected_at", &identity.occurrence_link.projected_at)
1279            .set("task_change_id", &identity.task_change_id)
1280            .set("occurrence_change_id", &identity.occurrence_change_id)
1281            .set(
1282                "task_field_version_seed",
1283                &identity.field_version_seeds.task,
1284            )
1285            .set(
1286                "occurrence_field_version_seed",
1287                &identity.field_version_seeds.occurrence,
1288            )
1289            .set("frequency", "daily")
1290            .set("interval", 1)
1291            .set("weekdays", "")
1292            .set("timezone", "UTC")
1293            .set("start_on", "2026-07-20")
1294            .set("available_local_time", "")
1295            .set("due_policy", "same_day")
1296            .into_value();
1297        let mut change = make_change_wire(
1298            op_type::PROJECT_RECURRENCE_OCCURRENCE,
1299            "recurrence_series",
1300            series_id.as_str(),
1301            payload,
1302        );
1303        change.change_id = identity.occurrence_change_id;
1304        change.field = Some("projection".to_string());
1305        change.created_at = identity.occurrence_link.projected_at;
1306        validate_pushed_change(&change).unwrap();
1307
1308        change.created_at = "2026-07-20T00:00:01Z".to_string();
1309        assert!(
1310            validate_pushed_change(&change)
1311                .unwrap_err()
1312                .to_string()
1313                .contains("recurrence-deterministic-mismatch")
1314        );
1315    }
1316
1317    #[test]
1318    fn request_pull_limit_has_default_and_bounds() {
1319        assert_eq!(request_pull_limit(None).unwrap(), MAX_PULL_BATCH);
1320        assert!(request_pull_limit(Some(MAX_PULL_BATCH)).is_ok());
1321        assert_eq!(
1322            request_pull_limit(Some(0)).unwrap_err().to_string(),
1323            "error sync-pull-limit-out-of-range min=1 max=512 got=0"
1324        );
1325        assert_eq!(
1326            request_pull_limit(Some(MAX_PULL_BATCH + 1))
1327                .unwrap_err()
1328                .to_string(),
1329            "error sync-pull-limit-out-of-range min=1 max=512 got=513"
1330        );
1331    }
1332
1333    #[test]
1334    fn request_envelope_rejects_negative_cursor_and_oversized_push_batch() {
1335        let request = SyncRequest {
1336            protocol_version: Some(SYNC_PROTOCOL_VERSION),
1337            client_id: "test-client".to_string(),
1338            after: -1,
1339            pull_limit: Some(MAX_PULL_BATCH),
1340            changes: Vec::new(),
1341        };
1342        assert_eq!(
1343            validate_sync_request_envelope(&request)
1344                .unwrap_err()
1345                .to_string(),
1346            "error sync-after-out-of-range min=0 got=-1"
1347        );
1348
1349        let request = SyncRequest {
1350            protocol_version: Some(SYNC_PROTOCOL_VERSION),
1351            client_id: "test-client".to_string(),
1352            after: 0,
1353            pull_limit: Some(MAX_PULL_BATCH),
1354            changes: vec![
1355                ChangeWire {
1356                    change_id: "AAAAAAAAAAAAAAA0".to_string(),
1357                    client_id: "client".to_string(),
1358                    local_seq: 1,
1359                    entity_type: "task".to_string(),
1360                    entity_id: "BBBBBBBBBBBBBBBB".to_string(),
1361                    field: None,
1362                    op_type: "create_task".to_string(),
1363                    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"}),
1364                    base_version: None,
1365                    created_at: "2026-01-01T00:00:00Z".to_string(),
1366                    server_seq: None,
1367                };
1368                MAX_PUSH_BATCH + 1
1369            ],
1370        };
1371        assert_eq!(
1372            validate_sync_request_envelope(&request)
1373                .unwrap_err()
1374                .to_string(),
1375            "error sync-push-too-large limit=256 got=257"
1376        );
1377    }
1378
1379    #[test]
1380    fn response_validation_respects_request_pull_limit() {
1381        let response = SyncResponse {
1382            protocol_version: SYNC_PROTOCOL_VERSION,
1383            cursor: 1,
1384            has_more: false,
1385            push_acks: vec![],
1386            changes: vec![
1387                ChangeWire {
1388                    change_id: "AAAAAAAAAAAAAAA1".to_string(),
1389                    client_id: "client".to_string(),
1390                    local_seq: 1,
1391                    entity_type: "task".to_string(),
1392                    entity_id: "BBBBBBBBBBBBBBBB".to_string(),
1393                    field: None,
1394                    op_type: "create_task".to_string(),
1395                    payload: serde_json::json!({
1396                        "title":"one",
1397                        "project_id":"0000000000000000",
1398                        "project_key":"app",
1399                        "project_name":"app",
1400                        "project_prefix":"APP",
1401                    }),
1402                    base_version: None,
1403                    created_at: "2026-01-01T00:00:00Z".to_string(),
1404                    server_seq: Some(1),
1405                };
1406                MAX_PULL_BATCH as usize + 1
1407            ],
1408        };
1409        assert_eq!(
1410            validate_sync_response_for_request(0, MAX_PULL_BATCH, &[], &response)
1411                .unwrap_err()
1412                .to_string(),
1413            "error invalid-sync-response pull-too-large limit=512 got=513"
1414        );
1415    }
1416
1417    #[test]
1418    fn constructed_create_task_payload_passes_wire_validation() {
1419        let ws = test_workspace();
1420        let payload = ChangePayload::workspace(&ws)
1421            .set("title", "test task")
1422            .set("description", "a description")
1423            .set("project_id", "1111111111111111")
1424            .set("project_key", "app")
1425            .set("project_name", "App")
1426            .set("project_prefix", "APP")
1427            .set("status", "inbox")
1428            .set("priority", "none")
1429            .set("created_at", "2026-06-01T00:00:00Z")
1430            .into_value();
1431        let change = make_change_wire(op_type::CREATE_TASK, "task", "BBBBBBBBBBBBBBBB", payload);
1432        validate_pushed_change(&change)
1433            .expect("create_task payload built with ChangePayload should be wire-valid");
1434    }
1435
1436    #[test]
1437    fn constructed_create_project_payload_passes_wire_validation() {
1438        let ws = test_workspace();
1439        let payload = ChangePayload::workspace(&ws)
1440            .set("key", "app")
1441            .set("name", "App")
1442            .set("prefix", "APP")
1443            .set("created_at", "2026-06-01T00:00:00Z")
1444            .into_value();
1445        let change = make_change_wire(
1446            op_type::CREATE_PROJECT,
1447            "project",
1448            "1111111111111111",
1449            payload,
1450        );
1451        validate_pushed_change(&change)
1452            .expect("create_project payload built with ChangePayload should be wire-valid");
1453    }
1454
1455    #[test]
1456    fn project_changes_reject_invalid_project_ids() {
1457        let ws = test_workspace();
1458        let project_payload = ChangePayload::workspace(&ws)
1459            .set("key", "app")
1460            .set("name", "App")
1461            .set("prefix", "APP")
1462            .set("created_at", "2026-06-01T00:00:00Z")
1463            .into_value();
1464        let create_project = make_change_wire(
1465            op_type::CREATE_PROJECT,
1466            "project",
1467            "invalid",
1468            project_payload,
1469        );
1470        assert_eq!(
1471            validate_pushed_change(&create_project)
1472                .unwrap_err()
1473                .to_string(),
1474            "error invalid-sync-change entity_id invalid-id"
1475        );
1476
1477        let task_payload = ChangePayload::workspace(&ws)
1478            .set("title", "test task")
1479            .set("project_id", "invalid")
1480            .set("project_key", "app")
1481            .set("project_name", "App")
1482            .set("project_prefix", "APP")
1483            .set("created_at", "2026-06-01T00:00:00Z")
1484            .into_value();
1485        let create_task = make_change_wire(
1486            op_type::CREATE_TASK,
1487            "task",
1488            "BBBBBBBBBBBBBBBB",
1489            task_payload,
1490        );
1491        assert_eq!(
1492            validate_pushed_change(&create_task)
1493                .unwrap_err()
1494                .to_string(),
1495            "error invalid-sync-change project_id invalid-id"
1496        );
1497
1498        let field_payload = ChangePayload::workspace(&ws)
1499            .set("value", "invalid")
1500            .set("project_id", "invalid")
1501            .set("project_key", "app")
1502            .set("project_name", "App")
1503            .set("project_prefix", "APP")
1504            .into_value();
1505        let mut set_field = make_change_wire(
1506            op_type::SET_FIELD,
1507            "task",
1508            "BBBBBBBBBBBBBBBB",
1509            field_payload,
1510        );
1511        set_field.field = Some("project".to_string());
1512        assert_eq!(
1513            validate_pushed_change(&set_field).unwrap_err().to_string(),
1514            "error invalid-sync-change project_id invalid-id"
1515        );
1516    }
1517
1518    #[test]
1519    fn constructed_label_add_payload_passes_wire_validation() {
1520        let ws = test_workspace();
1521        let payload = ChangePayload::workspace(&ws)
1522            .set("label", "bug")
1523            .into_value();
1524        let mut change = make_change_wire(op_type::LABEL_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1525        change.field = Some("labels".to_string());
1526        validate_pushed_change(&change)
1527            .expect("label_add payload built with ChangePayload should be wire-valid");
1528    }
1529
1530    #[test]
1531    fn constructed_dependency_add_payload_passes_wire_validation() {
1532        let ws = test_workspace();
1533        let payload = ChangePayload::workspace(&ws)
1534            .set("depends_on_task_id", "CCCCCCCCCCCCCCCC")
1535            .into_value();
1536        let mut change =
1537            make_change_wire(op_type::DEPENDENCY_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1538        change.field = Some("dependencies".to_string());
1539        validate_pushed_change(&change)
1540            .expect("dependency_add payload built with ChangePayload should be wire-valid");
1541    }
1542
1543    #[test]
1544    fn constructed_note_add_payload_passes_wire_validation() {
1545        let ws = test_workspace();
1546        let payload = ChangePayload::workspace(&ws)
1547            .set("note_id", "DDDDDDDDDDDDDDDD")
1548            .set("body", "note body")
1549            .set("created_at", "2026-06-01T00:00:00Z")
1550            .into_value();
1551        let mut change = make_change_wire(op_type::NOTE_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1552        change.field = Some("notes".to_string());
1553        validate_pushed_change(&change)
1554            .expect("note_add payload built with ChangePayload should be wire-valid");
1555    }
1556}
aven_core/sync/wire.rs

aven_core/sync/
wire.rs
