Sync 4: push endpoint POST /api/sync/push (LWW + history snapshot)
The core conflict-resolution step. Applies a batch of client changes,
additive + owner-scoped, with last-write-wins by client edit-time — and a
version-history snapshot on every overwrite so nothing is ever lost.
client_wins(client_edited_at, server_edited_at): apply iff client >= server;
a missing client time never overwrites a real server edit; a missing server
time (new row) yields. Notes compare against updated_at; labels gain an
updated_at (migration 0017, backfilled from created_at) as their LWW field.
Notes:
- upsert with a client-supplied id: create if absent, else LWW-apply the
full note state (title/body/color/kind/pins/trash/remind/position/items/
manual label_ids) with the same ripple as a web edit — derive_display_title,
_rewrite_links, _reconcile_tags (#tags), _rename_inbound_links. Overwriting
an existing title/body snapshots the old version into note_revisions first.
A resurrected tombstone clears purged_at.
- delete: purge tombstone (drop children + attachment files, clear content,
set purged_at), LWW-guarded so a newer server edit survives a stale delete.
Labels: upsert (create/rename/recolor) + delete (detach from notes, tombstone),
LWW-guarded; per-owner name-uniqueness clash on a different id is rejected
rather than raising.
Response: per-item {status: created|applied|kept|noop|rejected, sync_revision};
the client pulls afterward to converge. Whole-note semantics (client sends the
full state, not a partial patch).
Tests (DB-free): client_wins across all edit-time combinations; _parse_client_dt;
push auth-guard. Apply behavior + triggers operator-verified on deploy.
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:
@@ -0,0 +1,29 @@
|
||||
"""labels.updated_at (M8 sync hub, step 4 — LWW field for the label catalog)
|
||||
|
||||
Revision ID: 0017
|
||||
Revises: 0016
|
||||
Create Date: 2026-07-23
|
||||
|
||||
Labels need a last-edit timestamp so sync push can resolve label rename/recolor
|
||||
conflicts by last-write-wins (client edit-time vs this). Backfilled from created_at.
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision = "0017"
|
||||
down_revision = "0016"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column(
|
||||
"labels",
|
||||
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
||||
)
|
||||
# Seed existing rows' updated_at from their created_at (best available edit time).
|
||||
op.execute("UPDATE labels SET updated_at = created_at")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_column("labels", "updated_at")
|
||||
@@ -21,6 +21,11 @@ class Label(Base):
|
||||
name: Mapped[str] = mapped_column(Text(), nullable=False)
|
||||
color: Mapped[str] = mapped_column(Text(), nullable=False, default="default", server_default="default")
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
|
||||
# Last-edit time — the LWW comparison field for sync (a rename/recolor bumps it).
|
||||
# Sync-push overrides it with the winning client edit-time; web edits set server-now.
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
||||
)
|
||||
# Sync (M8): monotonic per-row revision (from sync_revision_seq via DB trigger) so a
|
||||
# label rename/recolor/merge/delete propagates to native clients independently of notes.
|
||||
sync_revision: Mapped[int | None] = mapped_column(BigInteger(), nullable=True)
|
||||
|
||||
+283
-3
@@ -11,19 +11,36 @@ from where it left off (since=0 = full initial sync).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from quart import Blueprint, g, jsonify, request
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy import delete as sa_delete
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from .auth import login_required
|
||||
from .config import Config
|
||||
from .db import session_scope
|
||||
from .models.label import Label
|
||||
from .models.label import Label, NoteLabel
|
||||
from .models.note import Note
|
||||
from .notes import _serialize_notes
|
||||
from .models.note_attachment import NoteAttachment
|
||||
from .models.note_item import NoteItem
|
||||
from .models.note_link import NoteLink
|
||||
from .models.note_revision import NoteRevision
|
||||
from .notes import (
|
||||
_reconcile_tags,
|
||||
_rename_inbound_links,
|
||||
_rewrite_links,
|
||||
_serialize_notes,
|
||||
derive_display_title,
|
||||
normalize_color,
|
||||
)
|
||||
|
||||
bp = Blueprint("sync", __name__, url_prefix="/api/sync")
|
||||
|
||||
DEFAULT_LIMIT = 500
|
||||
MAX_LIMIT = 1000
|
||||
MAX_PUSH = 1000 # per-batch change cap
|
||||
|
||||
|
||||
def _parse_since(raw: str | None) -> int:
|
||||
@@ -120,3 +137,266 @@ async def changes():
|
||||
"has_more": has_more,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# --- Push: apply client mutations (LWW by client edit-time, non-destructive) ---
|
||||
|
||||
|
||||
def _parse_client_dt(raw: object) -> datetime | None:
|
||||
if not isinstance(raw, str) or not raw:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def client_wins(client_edited_at: datetime | None, server_edited_at: datetime | None) -> bool:
|
||||
"""Last-write-wins: the client's version is applied iff its edit-time is at least
|
||||
the server's. A missing client time never overwrites a real server edit; a missing
|
||||
server time (new/unknown row) always yields to a present client edit."""
|
||||
if client_edited_at is None:
|
||||
return server_edited_at is None
|
||||
if server_edited_at is None:
|
||||
return True
|
||||
return client_edited_at >= server_edited_at
|
||||
|
||||
|
||||
def _assign_note_fields(note: Note, ch: dict) -> None:
|
||||
"""Overwrite a note's scalar fields from a client's FULL-state change (sync is
|
||||
whole-note, not a partial patch — the client sends its authoritative version)."""
|
||||
title = ch.get("title")
|
||||
note.title = (title or "").strip() or None if isinstance(title, str) else None
|
||||
note.body = ch["body"] if isinstance(ch.get("body"), str) else ""
|
||||
note.color = normalize_color(ch.get("color"))
|
||||
note.kind = ch["kind"] if ch.get("kind") in ("text", "list") else "text"
|
||||
note.pinned = bool(ch.get("pinned"))
|
||||
note.archived = bool(ch.get("archived"))
|
||||
if ch.get("trashed"):
|
||||
if note.deleted_at is None:
|
||||
note.deleted_at = datetime.now(timezone.utc)
|
||||
else:
|
||||
note.deleted_at = None
|
||||
note.remind_at = _parse_client_dt(ch.get("remind_at"))
|
||||
if isinstance(ch.get("position"), int):
|
||||
note.position = ch["position"]
|
||||
|
||||
|
||||
async def _apply_note_items(db, note: Note, ch: dict) -> None:
|
||||
"""Replace the note's checklist items with the client's (items sync inline)."""
|
||||
if note.kind != "list":
|
||||
await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
|
||||
return
|
||||
items = ch.get("items")
|
||||
if not isinstance(items, list):
|
||||
return
|
||||
await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
|
||||
for pos, it in enumerate(items):
|
||||
if not isinstance(it, dict):
|
||||
continue
|
||||
text = (it.get("text") or "").strip()
|
||||
if text:
|
||||
db.add(NoteItem(note_id=note.id, text=text, checked=bool(it.get("checked")), position=pos))
|
||||
|
||||
|
||||
async def _apply_note_manual_labels(db, note: Note, ch: dict) -> None:
|
||||
"""Set the note's MANUAL (picker) label memberships from client label_ids, leaving
|
||||
tag-sourced (via_tag) rows to _reconcile_tags. Only labels the caller owns count."""
|
||||
raw = ch.get("label_ids")
|
||||
if not isinstance(raw, list):
|
||||
return
|
||||
wanted: set = set()
|
||||
for r in raw:
|
||||
try:
|
||||
wanted.add(uuid.UUID(str(r)))
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
owned: set = set()
|
||||
if wanted:
|
||||
owned = set(
|
||||
(await db.scalars(select(Label.id).where(Label.owner_id == g.user_id, Label.id.in_(wanted)))).all()
|
||||
)
|
||||
existing = (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()
|
||||
attached = {r.label_id for r in existing}
|
||||
for r in existing:
|
||||
if not r.via_tag and r.label_id not in owned:
|
||||
await db.delete(r)
|
||||
for lid in owned:
|
||||
if lid not in attached:
|
||||
db.add(NoteLabel(note_id=note.id, label_id=lid, via_tag=False))
|
||||
|
||||
|
||||
async def _purge_note(db, note: Note, edited_at: datetime | None) -> None:
|
||||
"""Turn a note into a content-less tombstone: delete children (+ attachment files),
|
||||
clear content, set purged_at. Kept so offline clients learn it's gone."""
|
||||
atts = (await db.scalars(select(NoteAttachment).where(NoteAttachment.note_id == note.id))).all()
|
||||
for a in atts:
|
||||
try:
|
||||
(Config.media_root() / a.path).unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
await db.execute(sa_delete(NoteAttachment).where(NoteAttachment.note_id == note.id))
|
||||
await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
|
||||
await db.execute(sa_delete(NoteLabel).where(NoteLabel.note_id == note.id))
|
||||
await db.execute(sa_delete(NoteLink).where(NoteLink.source_id == note.id))
|
||||
note.title = None
|
||||
note.body = ""
|
||||
note.display_title = ""
|
||||
note.deleted_at = None
|
||||
note.remind_at = None
|
||||
note.purged_at = datetime.now(timezone.utc)
|
||||
if edited_at is not None:
|
||||
note.updated_at = edited_at
|
||||
|
||||
|
||||
async def _apply_note(db, ch: dict) -> dict:
|
||||
raw_id = ch.get("id")
|
||||
try:
|
||||
nid = uuid.UUID(str(raw_id))
|
||||
except (ValueError, TypeError):
|
||||
return {"id": raw_id, "entity": "note", "status": "rejected", "error": "invalid id"}
|
||||
op = ch.get("op", "upsert")
|
||||
edited_at = _parse_client_dt(ch.get("edited_at"))
|
||||
|
||||
note = await db.scalar(select(Note).where(Note.id == nid))
|
||||
if note is not None and note.owner_id != g.user_id:
|
||||
return {"id": str(nid), "entity": "note", "status": "rejected", "error": "not yours"}
|
||||
|
||||
if op == "delete":
|
||||
if note is None:
|
||||
return {"id": str(nid), "entity": "note", "status": "noop"}
|
||||
if not client_wins(edited_at, note.updated_at):
|
||||
return {"id": str(nid), "entity": "note", "status": "kept", "sync_revision": note.sync_revision}
|
||||
await _purge_note(db, note, edited_at)
|
||||
await db.flush()
|
||||
await db.refresh(note, ["sync_revision"])
|
||||
return {"id": str(nid), "entity": "note", "status": "applied", "sync_revision": note.sync_revision}
|
||||
|
||||
creating = note is None
|
||||
if creating:
|
||||
note = Note(id=nid, owner_id=g.user_id, body="", display_title="")
|
||||
created = _parse_client_dt(ch.get("created_at"))
|
||||
if created is not None:
|
||||
note.created_at = created
|
||||
db.add(note)
|
||||
elif not client_wins(edited_at, note.updated_at):
|
||||
return {"id": str(nid), "entity": "note", "status": "kept", "sync_revision": note.sync_revision}
|
||||
elif note.purged_at is not None:
|
||||
note.purged_at = None # client re-created/edited → clear the tombstone
|
||||
|
||||
old_title, old_body, old_display = note.title, note.body, note.display_title
|
||||
_assign_note_fields(note, ch)
|
||||
note.display_title = derive_display_title(note.title, note.body)
|
||||
if edited_at is not None:
|
||||
note.updated_at = edited_at
|
||||
# Non-destructive LWW: snapshot the overwritten server title+body into history.
|
||||
if not creating and (note.title != old_title or note.body != old_body):
|
||||
db.add(NoteRevision(note_id=note.id, title=old_title, body=old_body))
|
||||
await db.flush() # assign note.id before items/labels/links
|
||||
await _apply_note_items(db, note, ch)
|
||||
await _rewrite_links(db, note)
|
||||
await _reconcile_tags(db, note)
|
||||
await _apply_note_manual_labels(db, note, ch)
|
||||
new_display = note.display_title
|
||||
if old_display and new_display and old_display.strip().lower() != new_display.strip().lower():
|
||||
await _rename_inbound_links(db, note, old_display, new_display)
|
||||
await db.flush()
|
||||
await db.refresh(note, ["sync_revision"])
|
||||
return {
|
||||
"id": str(nid),
|
||||
"entity": "note",
|
||||
"status": "created" if creating else "applied",
|
||||
"sync_revision": note.sync_revision,
|
||||
}
|
||||
|
||||
|
||||
async def _apply_label(db, ch: dict) -> dict:
|
||||
raw_id = ch.get("id")
|
||||
try:
|
||||
lid = uuid.UUID(str(raw_id))
|
||||
except (ValueError, TypeError):
|
||||
return {"id": raw_id, "entity": "label", "status": "rejected", "error": "invalid id"}
|
||||
op = ch.get("op", "upsert")
|
||||
edited_at = _parse_client_dt(ch.get("edited_at"))
|
||||
|
||||
label = await db.scalar(select(Label).where(Label.id == lid))
|
||||
if label is not None and label.owner_id != g.user_id:
|
||||
return {"id": str(lid), "entity": "label", "status": "rejected", "error": "not yours"}
|
||||
|
||||
if op == "delete":
|
||||
if label is None:
|
||||
return {"id": str(lid), "entity": "label", "status": "noop"}
|
||||
if not client_wins(edited_at, label.updated_at):
|
||||
return {"id": str(lid), "entity": "label", "status": "kept", "sync_revision": label.sync_revision}
|
||||
await db.execute(sa_delete(NoteLabel).where(NoteLabel.label_id == label.id))
|
||||
label.purged_at = datetime.now(timezone.utc)
|
||||
if edited_at is not None:
|
||||
label.updated_at = edited_at
|
||||
await db.flush()
|
||||
await db.refresh(label, ["sync_revision"])
|
||||
return {"id": str(lid), "entity": "label", "status": "applied", "sync_revision": label.sync_revision}
|
||||
|
||||
name = (ch.get("name") or "").strip()
|
||||
creating = label is None
|
||||
if not creating and not client_wins(edited_at, label.updated_at):
|
||||
return {"id": str(lid), "entity": "label", "status": "kept", "sync_revision": label.sync_revision}
|
||||
# Names are unique per owner — a same-name clash on a DIFFERENT id can't be an insert.
|
||||
if name:
|
||||
clash = await db.scalar(
|
||||
select(Label.id).where(
|
||||
Label.owner_id == g.user_id, func.lower(Label.name) == name.lower(), Label.id != lid
|
||||
)
|
||||
)
|
||||
if clash is not None:
|
||||
return {"id": str(lid), "entity": "label", "status": "rejected", "error": "name in use"}
|
||||
|
||||
if creating:
|
||||
if not name:
|
||||
return {"id": str(lid), "entity": "label", "status": "rejected", "error": "name required"}
|
||||
label = Label(id=lid, owner_id=g.user_id, name=name, color=normalize_color(ch.get("color")))
|
||||
db.add(label)
|
||||
else:
|
||||
if label.purged_at is not None:
|
||||
label.purged_at = None
|
||||
if name:
|
||||
label.name = name
|
||||
label.color = normalize_color(ch.get("color"))
|
||||
if edited_at is not None:
|
||||
label.updated_at = edited_at
|
||||
await db.flush()
|
||||
await db.refresh(label, ["sync_revision"])
|
||||
return {
|
||||
"id": str(lid),
|
||||
"entity": "label",
|
||||
"status": "created" if creating else "applied",
|
||||
"sync_revision": label.sync_revision,
|
||||
}
|
||||
|
||||
|
||||
@bp.post("/push")
|
||||
@login_required
|
||||
async def push():
|
||||
"""Apply a batch of client changes. Additive + owner-scoped; LWW by client
|
||||
edit-time with a version-history snapshot on any overwrite (nothing is lost)."""
|
||||
body = await request.get_json(silent=True) or {}
|
||||
changes = body.get("changes")
|
||||
if not isinstance(changes, list):
|
||||
return jsonify({"error": "changes must be a list"}), 400
|
||||
if len(changes) > MAX_PUSH:
|
||||
return jsonify({"error": f"too many changes in one push (max {MAX_PUSH})"}), 400
|
||||
|
||||
results = []
|
||||
async with session_scope() as db:
|
||||
for ch in changes:
|
||||
if not isinstance(ch, dict):
|
||||
results.append({"status": "rejected", "error": "not an object"})
|
||||
continue
|
||||
entity = ch.get("entity")
|
||||
if entity == "note":
|
||||
results.append(await _apply_note(db, ch))
|
||||
elif entity == "label":
|
||||
results.append(await _apply_label(db, ch))
|
||||
else:
|
||||
results.append({"id": ch.get("id"), "status": "rejected", "error": "unknown entity"})
|
||||
await db.commit()
|
||||
return jsonify({"results": results})
|
||||
|
||||
+36
-1
@@ -1,7 +1,17 @@
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from thoughtsync.app import create_app
|
||||
from thoughtsync.sync import DEFAULT_LIMIT, MAX_LIMIT, _clamp_limit, _page_cursor, _parse_since
|
||||
from thoughtsync.sync import (
|
||||
DEFAULT_LIMIT,
|
||||
MAX_LIMIT,
|
||||
_clamp_limit,
|
||||
_page_cursor,
|
||||
_parse_client_dt,
|
||||
_parse_since,
|
||||
client_wins,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -15,6 +25,31 @@ async def test_changes_requires_auth(app):
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
async def test_push_requires_auth(app):
|
||||
client = app.test_client()
|
||||
resp = await client.post("/api/sync/push", json={"changes": []})
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_client_wins():
|
||||
older = datetime(2026, 7, 20, tzinfo=timezone.utc)
|
||||
newer = datetime(2026, 7, 22, tzinfo=timezone.utc)
|
||||
assert client_wins(newer, older) is True # newer client edit wins
|
||||
assert client_wins(older, newer) is False # older client edit loses (server kept)
|
||||
assert client_wins(older, older) is True # tie → client applies (idempotent)
|
||||
assert client_wins(None, older) is False # unknown client time can't overwrite a real edit
|
||||
assert client_wins(older, None) is True # new/unknown server side yields
|
||||
assert client_wins(None, None) is True
|
||||
|
||||
|
||||
def test_parse_client_dt():
|
||||
assert _parse_client_dt("2026-07-22T00:00:00Z").year == 2026
|
||||
assert _parse_client_dt("2026-07-22T00:00:00+00:00").tzinfo is not None
|
||||
assert _parse_client_dt("garbage") is None
|
||||
assert _parse_client_dt(None) is None
|
||||
assert _parse_client_dt(123) is None
|
||||
|
||||
|
||||
def test_parse_since():
|
||||
assert _parse_since(None) == 0
|
||||
assert _parse_since("42") == 42
|
||||
|
||||
Reference in New Issue
Block a user