From 8e40ea11889c8fc361378856f100ea2b7c31c6c0 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 22 Jul 2026 23:03:06 -0400 Subject: [PATCH] Sync 3: pull endpoint GET /api/sync/changes (M8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Delta pull for native clients: returns every note + label the caller owns whose sync_revision advanced past ?since=, ascending by revision, paginated (?limit, default 500 / max 1000), with the next cursor + has_more. since=0 is a full initial sync. Web app unaffected (new blueprint). Notes and labels share one revision sequence, so the cursor is a single watermark. _page_cursor() handles the two-stream paging: when either stream fills its page, it advances only to the SMALLER of the two page boundaries so nothing between the cursor and the next pull is skipped. Notes reuse _serialize_notes (items/labels/attachments inline) + sync_revision + purged_at (tombstone); labels carry name/color/purged_at/sync_revision. Returns ALL of the owner's notes regardless of state (active/archived/ trash/purged) — a client mirrors everything. Registered sync blueprint. Tests (DB-free): changes auth-guard; _parse_since / _clamp_limit validation; _page_cursor across empty / drained / one-full / both-full. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01FRgehjoz7Yv8LkUfADxACm --- src/thoughtsync/app.py | 2 + src/thoughtsync/sync.py | 122 ++++++++++++++++++++++++++++++++++++++++ tests/test_sync.py | 58 +++++++++++++++++++ 3 files changed, 182 insertions(+) create mode 100644 src/thoughtsync/sync.py create mode 100644 tests/test_sync.py diff --git a/src/thoughtsync/app.py b/src/thoughtsync/app.py index ebf6a6b..eae1b3c 100644 --- a/src/thoughtsync/app.py +++ b/src/thoughtsync/app.py @@ -16,6 +16,7 @@ from .labels import bp as labels_bp from .notes import bp as notes_bp from .settings import get_public_config, get_setting, load_or_create_secret_key from .settings_api import bp as settings_bp +from .sync import bp as sync_bp STATIC_DIR = os.path.join(os.path.dirname(__file__), "static") @@ -41,6 +42,7 @@ def create_app() -> Quart: app.register_blueprint(labels_bp) app.register_blueprint(graph_bp) app.register_blueprint(settings_bp) + app.register_blueprint(sync_bp) @app.before_serving async def _bootstrap() -> None: diff --git a/src/thoughtsync/sync.py b/src/thoughtsync/sync.py new file mode 100644 index 0000000..cbdb0d5 --- /dev/null +++ b/src/thoughtsync/sync.py @@ -0,0 +1,122 @@ +"""Delta-sync API for the local-first native clients (M8 sync hub). + +Pull: `GET /api/sync/changes?since=` returns every note + label the caller +owns whose sync_revision advanced past the cursor, newest-revision last, paginated. +Notes and labels both draw from ONE shared sequence (sync_revision_seq), so the +cursor is a single monotonic watermark across both entity types. + +The web app does NOT use this — it stays on the live REST API. This surface exists +purely so a native client can mirror the server into its local store and resume +from where it left off (since=0 = full initial sync). +""" +from __future__ import annotations + +from quart import Blueprint, g, jsonify, request +from sqlalchemy import select + +from .auth import login_required +from .db import session_scope +from .models.label import Label +from .models.note import Note +from .notes import _serialize_notes + +bp = Blueprint("sync", __name__, url_prefix="/api/sync") + +DEFAULT_LIMIT = 500 +MAX_LIMIT = 1000 + + +def _parse_since(raw: str | None) -> int: + """The pull cursor: a non-negative revision watermark. Bad/absent → 0 (full sync).""" + try: + return max(int(raw), 0) if raw is not None else 0 + except (ValueError, TypeError): + return 0 + + +def _clamp_limit(raw: str | None) -> int: + try: + return max(1, min(int(raw), MAX_LIMIT)) if raw is not None else DEFAULT_LIMIT + except (ValueError, TypeError): + return DEFAULT_LIMIT + + +def _page_cursor(note_revs: list[int], label_revs: list[int], since: int, limit: int) -> tuple[int, bool]: + """Compute the next cursor + has_more when paging TWO revision streams that share + one sequence. Each stream is fetched `rev > since ORDER BY rev LIMIT limit`. + + If either stream came back FULL (== limit) we're truncating, so the safe cursor is + the SMALLER of the two page boundaries — advancing only to where BOTH streams are + fully drained, so nothing between the cursor and the next pull is skipped. If + neither is full, everything ≤ max(returned) is drained. Both lists are ascending. + """ + boundaries = [] + if len(note_revs) == limit: + boundaries.append(note_revs[-1]) + if len(label_revs) == limit: + boundaries.append(label_revs[-1]) + if boundaries: + return min(boundaries), True + all_revs = note_revs + label_revs + return (max(all_revs) if all_revs else since), False + + +def _serialize_label_row(lb: Label) -> dict: + return { + "id": str(lb.id), + "name": lb.name, + "color": lb.color, + "sync_revision": lb.sync_revision, + "purged_at": lb.purged_at.isoformat() if lb.purged_at else None, + "created_at": lb.created_at.isoformat() if lb.created_at else None, + } + + +@bp.get("/changes") +@login_required +async def changes(): + since = _parse_since(request.args.get("since")) + limit = _clamp_limit(request.args.get("limit")) + async with session_scope() as db: + # ALL of the owner's notes/labels (any state — active/archived/trash/purged), + # since a client mirrors everything; ordered by the shared revision. + note_rows = ( + await db.scalars( + select(Note) + .where(Note.owner_id == g.user_id, Note.sync_revision > since) + .order_by(Note.sync_revision) + .limit(limit) + ) + ).all() + label_rows = ( + await db.scalars( + select(Label) + .where(Label.owner_id == g.user_id, Label.sync_revision > since) + .order_by(Label.sync_revision) + .limit(limit) + ) + ).all() + + cursor, has_more = _page_cursor( + [n.sync_revision for n in note_rows], + [lb.sync_revision for lb in label_rows], + since, + limit, + ) + # Trim each stream to the shared watermark so the two feeds stay aligned. + note_rows = [n for n in note_rows if n.sync_revision <= cursor] + label_rows = [lb for lb in label_rows if lb.sync_revision <= cursor] + + notes_out = await _serialize_notes(db, note_rows) + for data, n in zip(notes_out, note_rows): + data["sync_revision"] = n.sync_revision + data["purged_at"] = n.purged_at.isoformat() if n.purged_at else None + + return jsonify( + { + "notes": notes_out, + "labels": [_serialize_label_row(lb) for lb in label_rows], + "cursor": cursor, + "has_more": has_more, + } + ) diff --git a/tests/test_sync.py b/tests/test_sync.py new file mode 100644 index 0000000..1e7b315 --- /dev/null +++ b/tests/test_sync.py @@ -0,0 +1,58 @@ +import pytest + +from thoughtsync.app import create_app +from thoughtsync.sync import DEFAULT_LIMIT, MAX_LIMIT, _clamp_limit, _page_cursor, _parse_since + + +@pytest.fixture +def app(): + return create_app() + + +async def test_changes_requires_auth(app): + client = app.test_client() + resp = await client.get("/api/sync/changes") + assert resp.status_code == 401 + + +def test_parse_since(): + assert _parse_since(None) == 0 + assert _parse_since("42") == 42 + assert _parse_since("-5") == 0 # negative clamps to 0 + assert _parse_since("garbage") == 0 + + +def test_clamp_limit(): + assert _clamp_limit(None) == DEFAULT_LIMIT + assert _clamp_limit("10") == 10 + assert _clamp_limit("0") == 1 # floor of 1 + assert _clamp_limit("999999") == MAX_LIMIT + assert _clamp_limit("nope") == DEFAULT_LIMIT + + +def test_page_cursor_all_drained(): + # Neither stream is full → cursor is the max revision seen; nothing more to page. + cursor, more = _page_cursor([1, 3, 5], [2, 4], since=0, limit=500) + assert cursor == 5 + assert more is False + + +def test_page_cursor_empty(): + # No changes since the cursor → cursor stays put, no more pages. + cursor, more = _page_cursor([], [], since=7, limit=500) + assert cursor == 7 + assert more is False + + +def test_page_cursor_one_stream_full_advances_to_its_boundary(): + # Notes came back full (limit=3) → truncate at its boundary; later labels defer. + cursor, more = _page_cursor([1, 2, 3], [4, 5], since=0, limit=3) + assert cursor == 3 + assert more is True + + +def test_page_cursor_both_full_uses_min_boundary(): + # Both full → advance only to the SMALLER boundary so neither stream skips a gap. + cursor, more = _page_cursor([1, 2, 10], [3, 4, 5], since=0, limit=3) + assert cursor == 5 + assert more is True