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