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