Skip to main content

max / synckit

Cover the three pull variants over the mock server pull_filtered, pull_rich and pull_filtered_rich had no test of their own; only the shared pull_inner was exercised, through the base pull. Pin what each wrapper adds: that the filtered pair put tables and since into the POST body and omit both for a default filter, that the rich pair return PulledChange wrappers keeping the originating device_id and the server seq rather than the pulling device, and that a multi-page pull_rich resumes the second request from the cursor the first returned.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-23 22:19 UTC
Signed with PGP, not checked
Commit: d0b6e54ef9bbf806538f137631d2f289abc30107
Parent: fec69d9
1 file changed, +232 insertions, -0 deletions
M synckit-client/tests/integration/sync.rs +232
@@ -449,3 +449,235 @@
449 449 assert_eq!(a.table, b.table);
450 450 }
451 451 }
452 +
453 + // ── The three pull variants ──
454 + //
455 + // `pull_inner` is covered through the base `pull` above (cursor advance,
456 + // has_more, decryption, pagination). What each wrapper adds on top is the shape
457 + // of the request body it posts and the shape of what it hands back, so that is
458 + // what these pin.
459 +
460 + /// One encrypted change on the wire, as the server would return it.
461 + fn served_change(
462 + seq: i64,
463 + device_id: DeviceId,
464 + row_id: &str,
465 + key: &[u8; 32],
466 + plaintext: &serde_json::Value,
467 + ) -> serde_json::Value {
468 + let encrypted = synckit_client::crypto::encrypt_json(plaintext, key).unwrap();
469 + json!({
470 + "seq": seq,
471 + "device_id": device_id,
472 + "table": "tasks",
473 + "op": "INSERT",
474 + "row_id": row_id,
475 + "timestamp": "2025-06-01T12:00:00Z",
476 + "data": encrypted,
477 + })
478 + }
479 +
480 + #[tokio::test]
481 + async fn pull_filtered_puts_tables_and_since_in_the_body() {
482 + let kit = MockKit::start().await;
483 + let (client, key) = kit.keyed();
484 + let device_id = DeviceId::new(Uuid::new_v4());
485 + let plaintext = json!({"title": "filtered"});
486 +
487 + kit.post(PULL_PATH)
488 + .json(json!({
489 + "changes": [served_change(7, device_id, "row-1", &key, &plaintext)],
490 + "cursor": 7,
491 + "has_more": true,
492 + }))
493 + .await;
494 +
495 + let since = "2025-05-01T00:00:00Z"
496 + .parse::<chrono::DateTime<Utc>>()
497 + .unwrap();
498 + let filter = synckit_client::PullFilter {
499 + tables: Some(vec!["tasks".to_string(), "notes".to_string()]),
500 + since: Some(since),
501 + };
502 + let (changes, cursor, has_more) = client.pull_filtered(device_id, 3, filter).await.unwrap();
503 +
504 + assert_eq!(cursor, 7);
505 + assert!(has_more);
506 + assert_eq!(changes.len(), 1);
507 + assert_eq!(changes[0].row_id, "row-1");
508 + assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);
509 +
510 + let body = kit.body(PULL_PATH).await;
511 + assert_eq!(body["cursor"], 3);
512 + assert_eq!(body["device_id"], json!(device_id));
513 + assert_eq!(body["tables"], json!(["tasks", "notes"]));
514 + assert_eq!(
515 + body["since"]
516 + .as_str()
517 + .unwrap()
518 + .parse::<chrono::DateTime<Utc>>()
519 + .unwrap(),
520 + since
521 + );
522 + }
523 +
524 + #[tokio::test]
525 + async fn an_empty_filter_omits_both_fields_from_the_body() {
526 + let kit = MockKit::start().await;
527 + let (client, _key) = kit.keyed();
528 + let device_id = DeviceId::new(Uuid::new_v4());
529 +
530 + kit.post(PULL_PATH)
531 + .json(json!({"changes": [], "cursor": 0, "has_more": false}))
532 + .await;
533 +
534 + client
535 + .pull_filtered(device_id, 0, synckit_client::PullFilter::default())
536 + .await
537 + .unwrap();
538 +
539 + let body = kit.body(PULL_PATH).await;
540 + assert!(
541 + body.get("tables").is_none(),
542 + "an empty table list is not sent"
543 + );
544 + assert!(body.get("since").is_none(), "an absent since is not sent");
545 + }
546 +
547 + #[tokio::test]
548 + async fn pull_rich_keeps_device_id_and_seq() {
549 + let kit = MockKit::start().await;
550 + let (client, key) = kit.keyed();
551 + let origin = DeviceId::new(Uuid::new_v4());
552 + let me = DeviceId::new(Uuid::new_v4());
553 + let first = json!({"title": "one"});
554 + let second = json!({"title": "two"});
555 +
556 + kit.post(PULL_PATH)
557 + .json(json!({
558 + "changes": [
559 + served_change(41, origin, "row-a", &key, &first),
560 + served_change(42, origin, "row-b", &key, &second),
561 + ],
562 + "cursor": 42,
563 + "has_more": false,
564 + }))
565 + .await;
566 +
567 + let (changes, cursor, has_more) = client.pull_rich(me, 40).await.unwrap();
568 +
569 + assert_eq!(cursor, 42);
570 + assert!(!has_more);
571 + assert_eq!(changes.len(), 2);
572 + assert_eq!(changes[0].seq, 41);
573 + assert_eq!(changes[1].seq, 42);
574 + assert_eq!(changes[0].device_id, origin);
575 + assert_eq!(changes[1].device_id, origin);
576 + assert_ne!(
577 + changes[0].device_id, me,
578 + "the wrapper carries the originating device, not the puller"
579 + );
580 + assert_eq!(changes[0].entry.row_id, "row-a");
581 + assert_eq!(changes[0].entry.data.as_ref().unwrap(), &first);
582 + assert_eq!(changes[1].entry.data.as_ref().unwrap(), &second);
583 +
584 + let body = kit.body(PULL_PATH).await;
585 + assert_eq!(body["cursor"], 40);
586 + assert_eq!(body["device_id"], json!(me));
587 + assert!(body.get("tables").is_none(), "pull_rich sends no filter");
588 + }
589 +
590 + #[tokio::test]
591 + async fn pull_filtered_rich_carries_both_the_filter_and_the_metadata() {
592 + let kit = MockKit::start().await;
593 + let (client, key) = kit.keyed();
594 + let origin = DeviceId::new(Uuid::new_v4());
595 + let me = DeviceId::new(Uuid::new_v4());
596 + let plaintext = json!({"title": "both"});
597 +
598 + kit.post(PULL_PATH)
599 + .json(json!({
600 + "changes": [served_change(9, origin, "row-c", &key, &plaintext)],
601 + "cursor": 9,
602 + "has_more": false,
603 + }))
604 + .await;
605 +
606 + let since = "2025-04-02T03:04:05Z"
607 + .parse::<chrono::DateTime<Utc>>()
608 + .unwrap();
609 + let filter = synckit_client::PullFilter {
610 + tables: Some(vec!["tasks".to_string()]),
611 + since: Some(since),
612 + };
613 + let (changes, cursor, has_more) = client.pull_filtered_rich(me, 8, filter).await.unwrap();
614 +
615 + assert_eq!(cursor, 9);
616 + assert!(!has_more);
617 + assert_eq!(changes.len(), 1);
618 + assert_eq!(changes[0].seq, 9);
619 + assert_eq!(changes[0].device_id, origin);
620 + assert_eq!(changes[0].entry.row_id, "row-c");
621 + assert_eq!(changes[0].entry.data.as_ref().unwrap(), &plaintext);
622 +
623 + let body = kit.body(PULL_PATH).await;
624 + assert_eq!(body["cursor"], 8);
625 + assert_eq!(body["device_id"], json!(me));
626 + assert_eq!(body["tables"], json!(["tasks"]));
627 + assert_eq!(
628 + body["since"]
629 + .as_str()
630 + .unwrap()
631 + .parse::<chrono::DateTime<Utc>>()
632 + .unwrap(),
633 + since
634 + );
635 + }
636 +
637 + #[tokio::test]
638 + async fn pull_rich_drains_a_second_page() {
639 + let kit = MockKit::start().await;
640 + let (client, key) = kit.keyed();
641 + let origin = DeviceId::new(Uuid::new_v4());
642 + let me = DeviceId::new(Uuid::new_v4());
643 +
644 + kit.post(PULL_PATH)
645 + .once()
646 + .json(json!({
647 + "changes": [served_change(1, origin, "p1", &key, &json!({"n": 1}))],
648 + "cursor": 1,
649 + "has_more": true,
650 + }))
651 + .await;
652 + kit.post(PULL_PATH)
653 + .json(json!({
654 + "changes": [served_change(2, origin, "p2", &key, &json!({"n": 2}))],
655 + "cursor": 2,
656 + "has_more": false,
657 + }))
658 + .await;
659 +
660 + let mut cursor = 0;
661 + let mut collected: Vec<synckit_client::PulledChange> = Vec::new();
662 + loop {
663 + let (changes, next, has_more) = client.pull_rich(me, cursor).await.unwrap();
664 + collected.extend(changes);
665 + cursor = next;
666 + if !has_more {
667 + break;
668 + }
669 + }
670 +
671 + assert_eq!(cursor, 2);
672 + assert_eq!(collected.len(), 2, "the drain needed both round trips");
673 + assert_eq!(collected[0].seq, 1);
674 + assert_eq!(collected[1].seq, 2);
675 + assert_eq!(collected[0].entry.row_id, "p1");
676 + assert_eq!(collected[1].entry.row_id, "p2");
677 + assert_eq!(kit.hits(PULL_PATH).await, 2);
678 +
679 + // The second request resumes from the cursor the first returned.
680 + let bodies = kit.bodies("POST", PULL_PATH).await;
681 + assert_eq!(bodies[0]["cursor"], 0);
682 + assert_eq!(bodies[1]["cursor"], 1);
683 + }