1use std::collections::{HashMap, HashSet};
2
3use anyhow::{Context, Result, bail};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6
7use crate::attachments::validation::{
8 MAX_BLOB_BYTES, validate_alt_text, validate_blob_size, validate_dimensions, validate_filename,
9 validate_media_type,
10};
11use crate::change_log::op_type;
12use crate::choices::TaskSource;
13use crate::ids::{BASE32, ProjectId, WorkspaceId};
14use crate::task_fields::TaskField;
15
16pub const SYNC_PROTOCOL_VERSION: u32 = 9;
17pub fn sync_server_url_is_valid(server: &str) -> bool {
18 let Ok(url) = url::Url::parse(server) else {
19 return false;
20 };
21 matches!(url.scheme(), "http" | "https")
22 && url.host_str().is_some()
23 && url.username().is_empty()
24 && url.password().is_none()
25 && url.query().is_none()
26 && url.fragment().is_none()
27}
28pub const MAX_PUSH_BATCH: usize = 256;
29pub const MAX_PULL_BATCH: u32 = 512;
30pub const MAX_BLOB_TRANSFER_OBJECTS: usize = 16;
31pub const MAX_BLOB_TRANSFER_BYTES: u64 = 64 * 1024 * 1024;
32pub const DAEMON_SYNC_PAGE_BUDGET: usize = 8;
33pub const DAEMON_INCOMPLETE_RESCHEDULE_MS: u64 = 100;
34
35#[derive(Debug, Clone, Serialize, Deserialize)]
36pub struct ChangeWire {
37 pub change_id: String,
38 pub client_id: String,
39 pub local_seq: i64,
40 pub entity_type: String,
41 pub entity_id: String,
42 pub field: Option<String>,
43 pub op_type: String,
44 pub payload: Value,
45 pub base_version: Option<String>,
46 pub created_at: String,
47 pub server_seq: Option<i64>,
48}
49
50#[derive(Debug, Clone, Serialize, Deserialize)]
51pub struct PushAck {
52 pub change_id: String,
53 pub server_seq: i64,
54}
55
56#[derive(Debug)]
57pub struct ChangeRow {
58 pub change_id: String,
59 pub client_id: String,
60 pub local_seq: i64,
61 pub entity_type: String,
62 pub entity_id: String,
63 pub field: Option<String>,
64 pub op_type: String,
65 pub payload: String,
66 pub base_version: Option<String>,
67 pub created_at: String,
68 pub server_seq: Option<i64>,
69}
70
71impl ChangeRow {
72 pub fn into_wire(self) -> ChangeWire {
73 ChangeWire {
74 change_id: self.change_id,
75 client_id: self.client_id,
76 local_seq: self.local_seq,
77 entity_type: self.entity_type,
78 entity_id: self.entity_id,
79 field: self.field,
80 op_type: self.op_type,
81 payload: serde_json::from_str(&self.payload).unwrap_or(Value::Null),
82 base_version: self.base_version,
83 created_at: self.created_at,
84 server_seq: self.server_seq,
85 }
86 }
87}
88
89#[derive(Debug, Clone, Serialize, Deserialize)]
90pub struct SyncRequest {
91 #[serde(default)]
92 pub protocol_version: Option<u32>,
93 pub client_id: String,
94 pub after: i64,
95 #[serde(default)]
96 pub pull_limit: Option<u32>,
97 pub changes: Vec<ChangeWire>,
98}
99
100#[derive(Debug, Serialize, Deserialize)]
101pub struct SyncResponse {
102 pub protocol_version: u32,
103 pub cursor: i64,
104 pub has_more: bool,
105 #[serde(default)]
106 pub push_acks: Vec<PushAck>,
107 pub changes: Vec<ChangeWire>,
108}
109
110#[derive(Debug, Clone, Serialize, Deserialize)]
111pub struct BlobUploadContract {
112 pub workspace_id: String,
113 pub sha256: String,
114 pub byte_size: i64,
115 pub media_type: String,
116 pub width: i64,
117 pub height: i64,
118}
119
120#[derive(Debug, Serialize, Deserialize)]
121pub struct MissingBlobsRequest {
122 pub blobs: Vec<BlobUploadContract>,
123}
124
125#[derive(Debug, Serialize, Deserialize)]
126pub struct MissingBlobsResponse {
127 pub missing: Vec<String>,
128}
129
130pub fn validate_blob_contracts(blobs: &[BlobUploadContract]) -> Result<()> {
131 if blobs.len() > MAX_PUSH_BATCH {
132 bail!(
133 "error blob-batch-too-large limit={} got={}",
134 MAX_PUSH_BATCH,
135 blobs.len()
136 );
137 }
138 let mut seen = HashSet::with_capacity(blobs.len());
139 let mut hashes = HashSet::with_capacity(blobs.len());
140 for blob in blobs {
141 ensure_sync_id("workspace_id", &blob.workspace_id)?;
142 validate_sha256_for_sync(&blob.sha256)?;
143 validate_blob_size_for_sync(blob.byte_size)?;
144 map_attachment_validation(validate_media_type(&blob.media_type))?;
145 map_attachment_validation(validate_dimensions(Some(blob.width), Some(blob.height)))?;
146 if !seen.insert((blob.workspace_id.as_str(), blob.sha256.as_str())) {
147 bail!("error duplicate-blob-contract");
148 }
149 hashes.insert(blob.sha256.as_str());
150 }
151 if hashes.len() > MAX_BLOB_TRANSFER_OBJECTS {
152 bail!(
153 "error blob-batch-too-large limit={} got={}",
154 MAX_BLOB_TRANSFER_OBJECTS,
155 hashes.len()
156 );
157 }
158 Ok(())
159}
160
161pub fn validate_blob_hashes(hashes: &[String]) -> Result<()> {
162 if hashes.len() > MAX_BLOB_TRANSFER_OBJECTS {
163 bail!(
164 "error blob-batch-too-large limit={} got={}",
165 MAX_BLOB_TRANSFER_OBJECTS,
166 hashes.len()
167 );
168 }
169 let mut seen = HashSet::with_capacity(hashes.len());
170 for hash in hashes {
171 validate_sha256_for_sync(hash)?;
172 if !seen.insert(hash.as_str()) {
173 bail!("error duplicate-blob-hash");
174 }
175 }
176 Ok(())
177}
178
179#[derive(Debug)]
180struct SyncProtocolError {
181 client: u32,
182 server: u32,
183}
184
185impl std::fmt::Display for SyncProtocolError {
186 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
187 write!(
188 f,
189 "error sync-protocol-unsupported client={} server={}",
190 self.client, self.server
191 )
192 }
193}
194
195impl std::error::Error for SyncProtocolError {}
196
197fn sync_protocol_error(client: u32, server: u32) -> anyhow::Error {
198 anyhow::Error::new(SyncProtocolError { client, server })
199}
200
201pub fn validate_sync_protocol_version(client: u32, server: u32) -> Result<()> {
202 if client != server {
203 return Err(sync_protocol_error(client, server));
204 }
205 Ok(())
206}
207
208pub fn validate_sync_request_protocol_version(client: Option<u32>) -> Result<()> {
209 validate_sync_protocol_version(client.unwrap_or(0), SYNC_PROTOCOL_VERSION)
210}
211
212pub fn request_pull_limit(requested: Option<u32>) -> Result<u32> {
213 match requested {
214 None => Ok(MAX_PULL_BATCH),
215 Some(limit @ 1..=MAX_PULL_BATCH) => Ok(limit),
216 Some(limit) => {
217 bail!("error sync-pull-limit-out-of-range min=1 max={MAX_PULL_BATCH} got={limit}")
218 }
219 }
220}
221
222#[derive(Debug, Clone, Copy)]
223pub struct ValidatedSyncRequestEnvelope {
224 pub after: i64,
225 pub pull_limit: u32,
226 pub push_count: usize,
227}
228
229pub fn validate_sync_request_envelope(
230 request: &SyncRequest,
231) -> Result<ValidatedSyncRequestEnvelope> {
232 validate_sync_request_protocol_version(request.protocol_version)?;
233 validate_request_cursor(request.after)?;
234 validate_push_batch_size(request.changes.len())?;
235 Ok(ValidatedSyncRequestEnvelope {
236 after: request.after,
237 pull_limit: request_pull_limit(request.pull_limit)?,
238 push_count: request.changes.len(),
239 })
240}
241
242fn validate_request_cursor(after: i64) -> Result<()> {
243 if after < 0 {
244 bail!("error sync-after-out-of-range min=0 got={after}");
245 }
246 Ok(())
247}
248
249fn validate_push_batch_size(len: usize) -> Result<()> {
250 if len > MAX_PUSH_BATCH {
251 bail!("error sync-push-too-large limit={MAX_PUSH_BATCH} got={len}");
252 }
253 Ok(())
254}
255
256pub fn validate_sync_response_for_request(
257 after: i64,
258 pull_limit: u32,
259 request_change_ids: &[String],
260 response: &SyncResponse,
261) -> Result<()> {
262 validate_sync_protocol_version(SYNC_PROTOCOL_VERSION, response.protocol_version)?;
263 if response.changes.len() > pull_limit as usize {
264 bail!(
265 "error invalid-sync-response pull-too-large limit={} got={}",
266 pull_limit,
267 response.changes.len()
268 );
269 }
270 if response.cursor < after {
271 bail!(
272 "error invalid-sync-response cursor-regressed after={} cursor={}",
273 after,
274 response.cursor
275 );
276 }
277 validate_push_acks(request_change_ids, response)?;
278 validate_pull_page(after, pull_limit, response)?;
279 validate_push_pull_overlap(response)?;
280 Ok(())
281}
282
283#[derive(Debug, Clone, Copy, PartialEq, Eq)]
284enum ChangeDirection {
285 Pushed,
286 Pulled,
287}
288
289pub fn validate_pushed_change(change: &ChangeWire) -> Result<()> {
290 validate_change_shape(change, ChangeDirection::Pushed)
291}
292
293fn validate_pulled_change(change: &ChangeWire) -> Result<()> {
294 validate_change_shape(change, ChangeDirection::Pulled)
295}
296
297fn validate_change_shape(change: &ChangeWire, direction: ChangeDirection) -> Result<()> {
298 ensure_non_empty("change_id", &change.change_id)?;
299 ensure_non_empty("client_id", &change.client_id)?;
300 ensure_non_empty("entity_id", &change.entity_id)?;
301 ensure_non_empty("op_type", &change.op_type)?;
302 ensure_non_empty("entity_type", &change.entity_type)?;
303 if direction == ChangeDirection::Pushed {
304 ensure_sync_id("change_id", &change.change_id)?;
305 }
306 validate_change_server_seq(change, direction)?;
307 if !change.payload.is_object() {
308 bail!("error invalid-sync-change payload expected-object");
309 }
310
311 match change.op_type.as_str() {
312 op_type::CREATE_WORKSPACE => {
313 ensure_entity_type(change, "workspace")?;
314 ensure_sync_id("entity_id", &change.entity_id)?;
315 required_string_payload("key", &change.payload)?;
316 required_string_payload("name", &change.payload)?;
317 required_string_payload("created_at", &change.payload)?;
318 }
319 op_type::SET_WORKSPACE_FIELD => {
320 ensure_entity_type(change, "workspace")?;
321 ensure_sync_id("entity_id", &change.entity_id)?;
322 let field = change
323 .field
324 .as_deref()
325 .filter(|field| !field.trim().is_empty())
326 .context("error invalid-sync-change field missing")?;
327 if !matches!(field, "name" | "key") {
328 bail!("error invalid-sync-change field={field}");
329 }
330 required_string_payload("value", &change.payload)?;
331 }
332 op_type::CREATE_PROJECT => {
333 ensure_entity_type(change, "project")?;
334 ensure_project_id("entity_id", &change.entity_id)?;
335 optional_workspace_payload(&change.payload)?;
336 required_string_payload("key", &change.payload)?;
337 required_string_payload("name", &change.payload)?;
338 required_string_payload("prefix", &change.payload)?;
339 required_string_payload("created_at", &change.payload)?;
340 }
341 op_type::SET_PROJECT_METADATA => {
342 ensure_entity_type(change, "project")?;
343 ensure_project_id("entity_id", &change.entity_id)?;
344 optional_workspace_payload(&change.payload)?;
345 required_string_payload("key", &change.payload)?;
346 required_string_payload("name", &change.payload)?;
347 required_string_payload("prefix", &change.payload)?;
348 required_string_payload("updated_at", &change.payload)?;
349 }
350 op_type::CREATE_LABEL => {
351 ensure_entity_type(change, "label")?;
352 optional_workspace_payload(&change.payload)?;
353 required_string_payload("name", &change.payload)?;
354 required_string_payload("created_at", &change.payload)?;
355 }
356 op_type::CREATE_TASK => {
357 ensure_entity_type(change, "task")?;
358 ensure_sync_id("entity_id", &change.entity_id)?;
359 optional_workspace_payload(&change.payload)?;
360 required_string_payload("title", &change.payload)?;
361 let project_id = required_string_payload("project_id", &change.payload)?;
362 ensure_project_id("project_id", &project_id)?;
363 required_string_payload("project_key", &change.payload)?;
364 optional_string_payload("description", &change.payload)?;
365 required_string_payload("project_name", &change.payload)?;
366 required_string_payload("project_prefix", &change.payload)?;
367 if let Some(status) = optional_string_payload("status", &change.payload)? {
368 validate_sync_task_field_value(TaskField::Status, &status)?;
369 }
370 if let Some(priority) = optional_string_payload("priority", &change.payload)? {
371 validate_sync_task_field_value(TaskField::Priority, &priority)?;
372 }
373 if let Some(source) = optional_string_payload("source", &change.payload)? {
374 TaskSource::parse(&source)
375 .context("error invalid-sync-change invalid-task-source")?;
376 }
377 if let Some(available_at) = optional_string_payload("available_at", &change.payload)? {
378 validate_sync_task_field_value(TaskField::AvailableAt, &available_at)?;
379 }
380 if let Some(due_on) = optional_string_payload("due_on", &change.payload)? {
381 validate_sync_task_field_value(TaskField::DueOn, &due_on)?;
382 }
383 if let Some(is_epic) = optional_string_payload("is_epic", &change.payload)? {
384 validate_sync_task_field_value(TaskField::IsEpic, &is_epic)?;
385 }
386 optional_string_array_payload("labels", &change.payload)?;
387 optional_string_payload("created_at", &change.payload)?;
388 }
389 op_type::SET_FIELD | op_type::RESOLVE_FIELD => {
390 ensure_entity_type(change, "task")?;
391 ensure_sync_id("entity_id", &change.entity_id)?;
392 optional_workspace_payload(&change.payload)?;
393 let field = change
394 .field
395 .as_deref()
396 .filter(|field| !field.trim().is_empty())
397 .context("error invalid-sync-change field missing")?;
398 let task_field = TaskField::parse_for_sync(field)?;
399 let value = required_string_payload("value", &change.payload)?;
400 validate_sync_task_field_value(task_field, &value)?;
401 if task_field == TaskField::Project {
402 let project_id = required_string_payload("project_id", &change.payload)?;
403 ensure_project_id("project_id", &project_id)?;
404 if value != project_id {
405 bail!("error invalid-sync-change project-value-mismatch");
406 }
407 required_string_payload("project_key", &change.payload)?;
408 required_string_payload("project_name", &change.payload)?;
409 required_string_payload("project_prefix", &change.payload)?;
410 }
411 }
412 op_type::LABEL_ADD | op_type::LABEL_REMOVE => {
413 ensure_entity_type(change, "task")?;
414 ensure_sync_id("entity_id", &change.entity_id)?;
415 optional_workspace_payload(&change.payload)?;
416 required_string_payload("label", &change.payload)?;
417 }
418 op_type::NOTE_ADD => {
419 ensure_entity_type(change, "task")?;
420 ensure_sync_id("entity_id", &change.entity_id)?;
421 optional_workspace_payload(&change.payload)?;
422 let note_id = required_string_payload("note_id", &change.payload)?;
423 ensure_sync_id("note_id", ¬e_id)?;
424 required_string_payload("body", &change.payload)?;
425 required_string_payload("created_at", &change.payload)?;
426 }
427 op_type::DEPENDENCY_ADD | op_type::DEPENDENCY_REMOVE => {
428 ensure_entity_type(change, "task")?;
429 ensure_sync_id("entity_id", &change.entity_id)?;
430 required_workspace_payload(&change.payload)?;
431 let depends_on_task_id =
432 required_string_payload("depends_on_task_id", &change.payload)?;
433 ensure_sync_id("depends_on_task_id", &depends_on_task_id)?;
434 if change.entity_id == depends_on_task_id {
435 bail!("error invalid-sync-change dependency-self");
436 }
437 }
438 op_type::EPIC_LINK_ADD | op_type::EPIC_LINK_REMOVE => {
439 ensure_entity_type(change, "task")?;
440 ensure_sync_id("entity_id", &change.entity_id)?;
441 required_workspace_payload(&change.payload)?;
442 let epic_task_id = required_string_payload("epic_task_id", &change.payload)?;
443 ensure_sync_id("epic_task_id", &epic_task_id)?;
444 if change.entity_id == epic_task_id {
445 bail!("error invalid-sync-change epic-self");
446 }
447 if change.op_type == op_type::EPIC_LINK_ADD {
448 required_timestamp_payload("created_at", &change.payload)?;
449 }
450 }
451 op_type::PROJECT_DELETE => {
452 ensure_entity_type(change, "project")?;
453 ensure_project_id("entity_id", &change.entity_id)?;
454 required_workspace_payload(&change.payload)?;
455 required_timestamp_payload("deleted_at", &change.payload)?;
456 }
457 op_type::LABEL_DELETE => {
458 ensure_entity_type(change, "label")?;
459 required_workspace_payload(&change.payload)?;
460 let name = required_string_payload("name", &change.payload)?;
461 if name != change.entity_id {
462 bail!("error invalid-sync-change label-value-mismatch");
463 }
464 required_timestamp_payload("deleted_at", &change.payload)?;
465 }
466 op_type::NOTE_DELETE => {
467 ensure_entity_type(change, "task")?;
468 ensure_sync_id("entity_id", &change.entity_id)?;
469 if change.field.as_deref() != Some("notes") {
470 bail!("error invalid-sync-change field=notes");
471 }
472 required_workspace_payload(&change.payload)?;
473 let note_id = required_string_payload("note_id", &change.payload)?;
474 ensure_sync_id("note_id", ¬e_id)?;
475 required_timestamp_payload("deleted_at", &change.payload)?;
476 }
477 op_type::ATTACHMENT_ADD => validate_attachment_add_change(change)?,
478 op_type::ATTACHMENT_DELETE => validate_attachment_delete_change(change)?,
479 _ => bail!("error invalid-sync-change op_type={}", change.op_type),
480 }
481 Ok(())
482}
483
484fn validate_attachment_add_change(change: &ChangeWire) -> Result<()> {
485 ensure_entity_type(change, "task")?;
486 ensure_sync_id("entity_id", &change.entity_id)?;
487 ensure_attachment_field(change)?;
488 required_workspace_payload(&change.payload)?;
489 let attachment_id = required_string_payload("attachment_id", &change.payload)?;
490 ensure_sync_id("attachment_id", &attachment_id)?;
491 let sha256 = required_string_payload("sha256", &change.payload)?;
492 validate_sha256_for_sync(&sha256)?;
493 let byte_size = required_i64_payload("byte_size", &change.payload)?;
494 validate_blob_size_for_sync(byte_size)?;
495 let media_type = required_string_payload("media_type", &change.payload)?;
496 map_attachment_validation(validate_media_type(&media_type))?;
497 let filename = optional_string_payload("filename", &change.payload)?;
498 map_attachment_validation(validate_filename(filename.as_deref()))?;
499 let alt_text = optional_string_payload("alt_text", &change.payload)?;
500 map_attachment_validation(validate_alt_text(alt_text.as_deref()))?;
501 let width = optional_i64_payload("width", &change.payload)?;
502 let height = optional_i64_payload("height", &change.payload)?;
503 map_attachment_validation(validate_dimensions(width, height))?;
504 required_timestamp_payload("created_at", &change.payload)?;
505 Ok(())
506}
507
508fn validate_attachment_delete_change(change: &ChangeWire) -> Result<()> {
509 ensure_entity_type(change, "task")?;
510 ensure_sync_id("entity_id", &change.entity_id)?;
511 ensure_attachment_field(change)?;
512 required_workspace_payload(&change.payload)?;
513 let attachment_id = required_string_payload("attachment_id", &change.payload)?;
514 ensure_sync_id("attachment_id", &attachment_id)?;
515 required_timestamp_payload("deleted_at", &change.payload)?;
516 Ok(())
517}
518
519fn ensure_attachment_field(change: &ChangeWire) -> Result<()> {
520 if change.field.as_deref() != Some("attachments") {
521 bail!("error invalid-sync-change field=attachments");
522 }
523 Ok(())
524}
525
526fn validate_sha256_for_sync(value: &str) -> Result<()> {
527 if value.len() == 64
528 && value
529 .bytes()
530 .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
531 {
532 Ok(())
533 } else {
534 bail!("error invalid-sync-change invalid-sha256");
535 }
536}
537
538fn validate_blob_size_for_sync(byte_size: i64) -> Result<()> {
539 let Ok(bytes) = usize::try_from(byte_size) else {
540 bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
541 };
542 if bytes == 0 || bytes > MAX_BLOB_BYTES {
543 bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
544 }
545 map_attachment_validation(validate_blob_size(bytes))
546}
547
548fn map_attachment_validation(result: Result<()>) -> Result<()> {
549 result.map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
550}
551
552fn validate_change_server_seq(change: &ChangeWire, direction: ChangeDirection) -> Result<()> {
553 match direction {
554 ChangeDirection::Pushed if change.server_seq.is_some() => {
555 bail!("error invalid-sync-change server_seq client-supplied");
556 }
557 ChangeDirection::Pulled => match change.server_seq {
558 Some(server_seq) if server_seq > 0 => {}
559 Some(server_seq) => {
560 bail!("error invalid-sync-change server_seq={server_seq}");
561 }
562 None => bail!("error invalid-sync-change server_seq missing"),
563 },
564 ChangeDirection::Pushed => {}
565 }
566 Ok(())
567}
568
569fn validate_push_acks(request_change_ids: &[String], response: &SyncResponse) -> Result<()> {
570 if response.push_acks.len() != request_change_ids.len() {
571 bail!(
572 "error invalid-sync-response push-ack-count expected={} got={}",
573 request_change_ids.len(),
574 response.push_acks.len()
575 );
576 }
577 let expected = request_change_ids
578 .iter()
579 .map(String::as_str)
580 .collect::<HashSet<_>>();
581 let mut seen = HashSet::with_capacity(response.push_acks.len());
582 for ack in &response.push_acks {
583 if !expected.contains(ack.change_id.as_str()) {
584 bail!(
585 "error invalid-sync-response unexpected-push-ack change_id={}",
586 ack.change_id
587 );
588 }
589 if ack.server_seq <= 0 {
590 bail!(
591 "error invalid-sync-response push-ack-server-seq change_id={} server_seq={}",
592 ack.change_id,
593 ack.server_seq
594 );
595 }
596 if !seen.insert(ack.change_id.as_str()) {
597 bail!(
598 "error invalid-sync-response duplicate-push-ack change_id={}",
599 ack.change_id
600 );
601 }
602 }
603 Ok(())
604}
605
606fn validate_pull_page(after: i64, pull_limit: u32, response: &SyncResponse) -> Result<()> {
607 let mut previous = after;
608 let mut change_ids = HashSet::with_capacity(response.changes.len());
609 for change in &response.changes {
610 validate_pulled_change(change)?;
611 if !change_ids.insert(&change.change_id) {
612 bail!(
613 "error invalid-sync-response duplicate-pull-change change_id={}",
614 change.change_id
615 );
616 }
617 let server_seq = change.server_seq.with_context(|| {
618 format!(
619 "error invalid-sync-response missing-server-seq change_id={}",
620 change.change_id
621 )
622 })?;
623 if server_seq <= previous {
624 bail!(
625 "error invalid-sync-response server-seq-order previous={} server_seq={}",
626 previous,
627 server_seq
628 );
629 }
630 previous = server_seq;
631 }
632 let expected_cursor = response
633 .changes
634 .last()
635 .and_then(|change| change.server_seq)
636 .unwrap_or(after);
637 if response.cursor != expected_cursor {
638 bail!(
639 "error invalid-sync-response cursor-mismatch expected={} got={}",
640 expected_cursor,
641 response.cursor
642 );
643 }
644 if response.has_more && response.changes.len() < pull_limit as usize {
645 bail!(
646 "error invalid-sync-response has-more-short-page returned={} limit={}",
647 response.changes.len(),
648 pull_limit
649 );
650 }
651 Ok(())
652}
653
654fn validate_push_pull_overlap(response: &SyncResponse) -> Result<()> {
655 let acked = response
656 .push_acks
657 .iter()
658 .map(|ack| (ack.change_id.as_str(), ack.server_seq))
659 .collect::<HashMap<_, _>>();
660 for change in &response.changes {
661 if let Some(acked_server_seq) = acked.get(change.change_id.as_str()) {
662 let Some(pull_server_seq) = change.server_seq else {
663 continue;
664 };
665 if *acked_server_seq != pull_server_seq {
666 bail!(
667 "error invalid-sync-response push-pull-server-seq-mismatch change_id={} ack={} pull={}",
668 change.change_id,
669 acked_server_seq,
670 pull_server_seq
671 );
672 }
673 }
674 }
675 Ok(())
676}
677
678fn validate_sync_task_field_value(field: TaskField, value: &str) -> Result<()> {
679 field
680 .validate_value(value)
681 .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
682}
683
684fn ensure_entity_type(change: &ChangeWire, expected: &str) -> Result<()> {
685 if change.entity_type == expected {
686 Ok(())
687 } else {
688 bail!(
689 "error invalid-sync-change op_type={} entity_type={} expected={}",
690 change.op_type,
691 change.entity_type,
692 expected
693 )
694 }
695}
696
697fn ensure_non_empty(name: &str, value: &str) -> Result<()> {
698 if value.trim().is_empty() {
699 bail!("error invalid-sync-change {name} empty");
700 }
701 Ok(())
702}
703
704fn ensure_sync_id(name: &str, value: &str) -> Result<()> {
705 if value.len() == 16 && value.bytes().all(|byte| BASE32.contains(&byte)) {
706 Ok(())
707 } else {
708 bail!("error invalid-sync-change {name} invalid-id");
709 }
710}
711
712fn ensure_project_id(name: &str, value: &str) -> Result<()> {
713 if value.parse::<ProjectId>().is_ok() {
714 Ok(())
715 } else {
716 bail!("error invalid-sync-change {name} invalid-id");
717 }
718}
719
720fn required_string_payload(key: &str, payload: &Value) -> Result<String> {
721 payload
722 .get(key)
723 .and_then(Value::as_str)
724 .map(str::to_string)
725 .with_context(|| format!("error invalid-sync-change payload.{key} missing"))
726}
727
728fn required_i64_payload(key: &str, payload: &Value) -> Result<i64> {
729 match payload.get(key) {
730 Some(Value::Number(value)) => value
731 .as_i64()
732 .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
733 Some(Value::Null) | None => bail!("error invalid-sync-change payload.{key} missing"),
734 Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
735 }
736}
737
738fn required_timestamp_payload(key: &str, payload: &Value) -> Result<String> {
739 let value = required_string_payload(key, payload)?;
740 if value.len() == 20
741 && value.as_bytes()[4] == b'-'
742 && value.as_bytes()[7] == b'-'
743 && value.as_bytes()[10] == b'T'
744 && value.as_bytes()[13] == b':'
745 && value.as_bytes()[16] == b':'
746 && value.as_bytes()[19] == b'Z'
747 && value
748 .bytes()
749 .enumerate()
750 .all(|(idx, byte)| matches!(idx, 4 | 7 | 10 | 13 | 16 | 19) || byte.is_ascii_digit())
751 {
752 Ok(value)
753 } else {
754 bail!("error invalid-sync-change payload.{key} invalid-timestamp");
755 }
756}
757
758fn required_workspace_payload(payload: &Value) -> Result<()> {
759 let id = required_string_payload("workspace_id", payload)?;
760 if id.parse::<WorkspaceId>().is_err() {
761 bail!("error invalid-sync-change workspace_id invalid-id");
762 }
763 required_string_payload("workspace_key", payload)?;
764 Ok(())
765}
766
767fn optional_workspace_payload(payload: &Value) -> Result<()> {
768 if payload.get("workspace_id").is_none() && payload.get("workspace_key").is_none() {
769 return Ok(());
770 }
771 required_workspace_payload(payload)
772}
773
774fn optional_string_payload(key: &str, payload: &Value) -> Result<Option<String>> {
775 match payload.get(key) {
776 Some(Value::String(value)) => Ok(Some(value.clone())),
777 Some(Value::Null) | None => Ok(None),
778 Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
779 }
780}
781
782fn optional_i64_payload(key: &str, payload: &Value) -> Result<Option<i64>> {
783 match payload.get(key) {
784 Some(Value::Number(value)) => value
785 .as_i64()
786 .map(Some)
787 .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
788 Some(Value::Null) | None => Ok(None),
789 Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
790 }
791}
792
793fn optional_string_array_payload(key: &str, payload: &Value) -> Result<()> {
794 match payload.get(key) {
795 Some(Value::Array(values))
796 if values
797 .iter()
798 .all(|value| value.as_str().is_some_and(|value| !value.trim().is_empty())) =>
799 {
800 Ok(())
801 }
802 Some(Value::Null) | None => Ok(()),
803 Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
804 }
805}
806
807#[cfg(test)]
808mod tests {
809 use super::*;
810 use crate::change_log::{ChangePayload, op_type};
811 use crate::workspaces::Workspace;
812
813 fn test_workspace() -> Workspace {
814 Workspace {
815 id: "0000000000000000".parse().unwrap(),
816 key: "default".to_string(),
817 name: "default".to_string(),
818 }
819 }
820
821 fn make_change_wire(
822 op_type: &str,
823 entity_type: &str,
824 entity_id: &str,
825 payload: serde_json::Value,
826 ) -> ChangeWire {
827 ChangeWire {
828 change_id: "AAAAAAAAAAAAAAA0".to_string(),
829 client_id: "client".to_string(),
830 local_seq: 1,
831 entity_type: entity_type.to_string(),
832 entity_id: entity_id.to_string(),
833 field: None,
834 op_type: op_type.to_string(),
835 payload,
836 base_version: None,
837 created_at: "2026-06-01T00:00:00Z".to_string(),
838 server_seq: None,
839 }
840 }
841
842 #[test]
843 fn request_pull_limit_has_default_and_bounds() {
844 assert_eq!(request_pull_limit(None).unwrap(), MAX_PULL_BATCH);
845 assert!(request_pull_limit(Some(MAX_PULL_BATCH)).is_ok());
846 assert_eq!(
847 request_pull_limit(Some(0)).unwrap_err().to_string(),
848 "error sync-pull-limit-out-of-range min=1 max=512 got=0"
849 );
850 assert_eq!(
851 request_pull_limit(Some(MAX_PULL_BATCH + 1))
852 .unwrap_err()
853 .to_string(),
854 "error sync-pull-limit-out-of-range min=1 max=512 got=513"
855 );
856 }
857
858 #[test]
859 fn request_envelope_rejects_negative_cursor_and_oversized_push_batch() {
860 let request = SyncRequest {
861 protocol_version: Some(SYNC_PROTOCOL_VERSION),
862 client_id: "test-client".to_string(),
863 after: -1,
864 pull_limit: Some(MAX_PULL_BATCH),
865 changes: Vec::new(),
866 };
867 assert_eq!(
868 validate_sync_request_envelope(&request)
869 .unwrap_err()
870 .to_string(),
871 "error sync-after-out-of-range min=0 got=-1"
872 );
873
874 let request = SyncRequest {
875 protocol_version: Some(SYNC_PROTOCOL_VERSION),
876 client_id: "test-client".to_string(),
877 after: 0,
878 pull_limit: Some(MAX_PULL_BATCH),
879 changes: vec![
880 ChangeWire {
881 change_id: "AAAAAAAAAAAAAAA0".to_string(),
882 client_id: "client".to_string(),
883 local_seq: 1,
884 entity_type: "task".to_string(),
885 entity_id: "BBBBBBBBBBBBBBBB".to_string(),
886 field: None,
887 op_type: "create_task".to_string(),
888 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"}),
889 base_version: None,
890 created_at: "2026-01-01T00:00:00Z".to_string(),
891 server_seq: None,
892 };
893 MAX_PUSH_BATCH + 1
894 ],
895 };
896 assert_eq!(
897 validate_sync_request_envelope(&request)
898 .unwrap_err()
899 .to_string(),
900 "error sync-push-too-large limit=256 got=257"
901 );
902 }
903
904 #[test]
905 fn response_validation_respects_request_pull_limit() {
906 let response = SyncResponse {
907 protocol_version: SYNC_PROTOCOL_VERSION,
908 cursor: 1,
909 has_more: false,
910 push_acks: vec![],
911 changes: vec![
912 ChangeWire {
913 change_id: "AAAAAAAAAAAAAAA1".to_string(),
914 client_id: "client".to_string(),
915 local_seq: 1,
916 entity_type: "task".to_string(),
917 entity_id: "BBBBBBBBBBBBBBBB".to_string(),
918 field: None,
919 op_type: "create_task".to_string(),
920 payload: serde_json::json!({
921 "title":"one",
922 "project_id":"0000000000000000",
923 "project_key":"app",
924 "project_name":"app",
925 "project_prefix":"APP",
926 }),
927 base_version: None,
928 created_at: "2026-01-01T00:00:00Z".to_string(),
929 server_seq: Some(1),
930 };
931 MAX_PULL_BATCH as usize + 1
932 ],
933 };
934 assert_eq!(
935 validate_sync_response_for_request(0, MAX_PULL_BATCH, &[], &response)
936 .unwrap_err()
937 .to_string(),
938 "error invalid-sync-response pull-too-large limit=512 got=513"
939 );
940 }
941
942 #[test]
943 fn constructed_create_task_payload_passes_wire_validation() {
944 let ws = test_workspace();
945 let payload = ChangePayload::workspace(&ws)
946 .set("title", "test task")
947 .set("description", "a description")
948 .set("project_id", "1111111111111111")
949 .set("project_key", "app")
950 .set("project_name", "App")
951 .set("project_prefix", "APP")
952 .set("status", "inbox")
953 .set("priority", "none")
954 .set("created_at", "2026-06-01T00:00:00Z")
955 .into_value();
956 let change = make_change_wire(op_type::CREATE_TASK, "task", "BBBBBBBBBBBBBBBB", payload);
957 validate_pushed_change(&change)
958 .expect("create_task payload built with ChangePayload should be wire-valid");
959 }
960
961 #[test]
962 fn constructed_create_project_payload_passes_wire_validation() {
963 let ws = test_workspace();
964 let payload = ChangePayload::workspace(&ws)
965 .set("key", "app")
966 .set("name", "App")
967 .set("prefix", "APP")
968 .set("created_at", "2026-06-01T00:00:00Z")
969 .into_value();
970 let change = make_change_wire(
971 op_type::CREATE_PROJECT,
972 "project",
973 "1111111111111111",
974 payload,
975 );
976 validate_pushed_change(&change)
977 .expect("create_project payload built with ChangePayload should be wire-valid");
978 }
979
980 #[test]
981 fn project_changes_reject_invalid_project_ids() {
982 let ws = test_workspace();
983 let project_payload = ChangePayload::workspace(&ws)
984 .set("key", "app")
985 .set("name", "App")
986 .set("prefix", "APP")
987 .set("created_at", "2026-06-01T00:00:00Z")
988 .into_value();
989 let create_project = make_change_wire(
990 op_type::CREATE_PROJECT,
991 "project",
992 "invalid",
993 project_payload,
994 );
995 assert_eq!(
996 validate_pushed_change(&create_project)
997 .unwrap_err()
998 .to_string(),
999 "error invalid-sync-change entity_id invalid-id"
1000 );
1001
1002 let task_payload = ChangePayload::workspace(&ws)
1003 .set("title", "test task")
1004 .set("project_id", "invalid")
1005 .set("project_key", "app")
1006 .set("project_name", "App")
1007 .set("project_prefix", "APP")
1008 .set("created_at", "2026-06-01T00:00:00Z")
1009 .into_value();
1010 let create_task = make_change_wire(
1011 op_type::CREATE_TASK,
1012 "task",
1013 "BBBBBBBBBBBBBBBB",
1014 task_payload,
1015 );
1016 assert_eq!(
1017 validate_pushed_change(&create_task)
1018 .unwrap_err()
1019 .to_string(),
1020 "error invalid-sync-change project_id invalid-id"
1021 );
1022
1023 let field_payload = ChangePayload::workspace(&ws)
1024 .set("value", "invalid")
1025 .set("project_id", "invalid")
1026 .set("project_key", "app")
1027 .set("project_name", "App")
1028 .set("project_prefix", "APP")
1029 .into_value();
1030 let mut set_field = make_change_wire(
1031 op_type::SET_FIELD,
1032 "task",
1033 "BBBBBBBBBBBBBBBB",
1034 field_payload,
1035 );
1036 set_field.field = Some("project".to_string());
1037 assert_eq!(
1038 validate_pushed_change(&set_field).unwrap_err().to_string(),
1039 "error invalid-sync-change project_id invalid-id"
1040 );
1041 }
1042
1043 #[test]
1044 fn constructed_label_add_payload_passes_wire_validation() {
1045 let ws = test_workspace();
1046 let payload = ChangePayload::workspace(&ws)
1047 .set("label", "bug")
1048 .into_value();
1049 let mut change = make_change_wire(op_type::LABEL_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1050 change.field = Some("labels".to_string());
1051 validate_pushed_change(&change)
1052 .expect("label_add payload built with ChangePayload should be wire-valid");
1053 }
1054
1055 #[test]
1056 fn constructed_dependency_add_payload_passes_wire_validation() {
1057 let ws = test_workspace();
1058 let payload = ChangePayload::workspace(&ws)
1059 .set("depends_on_task_id", "CCCCCCCCCCCCCCCC")
1060 .into_value();
1061 let mut change =
1062 make_change_wire(op_type::DEPENDENCY_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1063 change.field = Some("dependencies".to_string());
1064 validate_pushed_change(&change)
1065 .expect("dependency_add payload built with ChangePayload should be wire-valid");
1066 }
1067
1068 #[test]
1069 fn constructed_note_add_payload_passes_wire_validation() {
1070 let ws = test_workspace();
1071 let payload = ChangePayload::workspace(&ws)
1072 .set("note_id", "DDDDDDDDDDDDDDDD")
1073 .set("body", "note body")
1074 .set("created_at", "2026-06-01T00:00:00Z")
1075 .into_value();
1076 let mut change = make_change_wire(op_type::NOTE_ADD, "task", "BBBBBBBBBBBBBBBB", payload);
1077 change.field = Some("notes".to_string());
1078 validate_pushed_change(&change)
1079 .expect("note_add payload built with ChangePayload should be wire-valid");
1080 }
1081}