Sync 3: pull endpoint GET /api/sync/changes (M8)
Delta pull for native clients: returns every note + label the caller owns whose sync_revision advanced past ?since=<cursor>, 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) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FRgehjoz7Yv8LkUfADxACm
This commit is contained in:
@@ -16,6 +16,7 @@ from .labels import bp as labels_bp
|
|||||||
from .notes import bp as notes_bp
|
from .notes import bp as notes_bp
|
||||||
from .settings import get_public_config, get_setting, load_or_create_secret_key
|
from .settings import get_public_config, get_setting, load_or_create_secret_key
|
||||||
from .settings_api import bp as settings_bp
|
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")
|
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(labels_bp)
|
||||||
app.register_blueprint(graph_bp)
|
app.register_blueprint(graph_bp)
|
||||||
app.register_blueprint(settings_bp)
|
app.register_blueprint(settings_bp)
|
||||||
|
app.register_blueprint(sync_bp)
|
||||||
|
|
||||||
@app.before_serving
|
@app.before_serving
|
||||||
async def _bootstrap() -> None:
|
async def _bootstrap() -> None:
|
||||||
|
|||||||
@@ -0,0 +1,122 @@
|
|||||||
|
"""Delta-sync API for the local-first native clients (M8 sync hub).
|
||||||
|
|
||||||
|
Pull: `GET /api/sync/changes?since=<cursor>` 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,
|
||||||
|
}
|
||||||
|
)
|
||||||
@@ -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
|
||||||
Reference in New Issue
Block a user