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;
48pub 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", ¬e_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", ¬e_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", ¬e_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}