Skip to main content

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