sync: shared notes in the feed, revocations, and text pushes from an edit share

The change feed answers `?shares=1` with every note the caller can see, each
saying how it is held, plus a `revoked` list of notes that left them. Granting
a share moves the note past the recipient's cursor; ending one, or the owner
deleting the note, leaves a revocation on the same cursor. A recipient at edit
may push the note's text, and nothing else. Protocol 6, feature `shares`;
opt-in, so the floor stays at 3.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_integration.py

@@ -51,7 +51,7 @@ pytestmark = pytest.mark.integration
 
 # Every table the tests touch, child-first so FKs never block the truncate.
 # RESTART IDENTITY + CASCADE keeps this honest if a table gains children later.
-_TABLES = "notes, note_revisions, note_labels, note_link_previews, labels, shares, invites, password_resets, users"
+_TABLES = "notes, note_revisions, note_labels, note_link_previews, labels, shares, share_revocations, invites, password_resets, users"
 
 
 @pytest_asyncio.fixture
@@ -1343,6 +1343,100 @@ async def test_a_share_names_someone_else_on_this_instance(app_client, db):
     assert (await app_client.delete(f"/api/notes/{other}/shares/{sid}")).status_code == 404
 
 
+# --- Shared notes over sync (#5175) -------------------------------------------
+
+
+async def _feed(client, since: int = 0, shares: bool = True) -> dict:
+    query = f"since={since}" + ("&shares=1" if shares else "")
+    resp = await client.get(f"/api/sync/changes?{query}")
+    assert resp.status_code == 200
+    return await resp.get_json()
+
+
+def _feed_note(page: dict, nid: str) -> dict | None:
+    return next((n for n in page["notes"] if n["id"] == nid), None)
+
+
+async def test_a_share_reaches_the_recipients_feed_and_a_revoke_takes_it_back(app_client, db):
+    recipient, stranger, people = await _three_people(app_client)
+    nid = await _owners_note(app_client)
+    start = (await _feed(recipient))["cursor"]
+
+    share_id = (await (await _share(app_client, nid, people["recipient"])).get_json())["shares"][0]["id"]
+    granted = await _feed(recipient, start)
+    held = _feed_note(granted, nid)
+    assert held is not None, "the grant moved the note past the recipient's cursor"
+    assert (held["permission"], held["shared_by"]["display_name"], held["labels"]) == ("view", "owner", [])
+    assert granted["revoked"] == []
+    # A client that never asked for shares gets only its own notes, as before.
+    assert _feed_note(await _feed(recipient, shares=False), nid) is None
+    assert "revoked" not in await _feed(recipient, shares=False)
+    assert _feed_note(await _feed(stranger), nid) is None
+
+    # The owner's devices learn the note is shared.
+    mine = _feed_note(await _feed(app_client), nid)
+    assert (mine["permission"], mine["shared"], [lb["name"] for lb in mine["labels"]]) == ("owner", True, ["idea"])
+
+    await app_client.delete(f"/api/notes/{nid}/shares/{share_id}")
+    revoked = await _feed(recipient, granted["cursor"])
+    assert revoked["revoked"] == [nid]
+    assert _feed_note(revoked, nid) is None
+    assert _feed_note(await _feed(app_client), nid)["shared"] is False
+    assert (await _feed(stranger))["revoked"] == []
+
+    # Shared again: a device that never heard of the revocation isn't told to delete it.
+    await _share(app_client, nid, people["recipient"], "edit")
+    fresh = await _feed(recipient, start)
+    assert fresh["revoked"] == []
+    assert _feed_note(fresh, nid)["permission"] == "edit"
+
+
+async def test_an_edit_share_pushes_text_and_nothing_else(app_client, db):
+    recipient, stranger, people = await _three_people(app_client)
+    nid = await _owners_note(app_client, "first line")
+    await _share(app_client, nid, people["recipient"], "view")
+
+    def change(**fields) -> dict:
+        return {"changes": [{"entity": "note", "id": nid, "op": "upsert", "edited_at": "2099-01-01T00:00:00Z", **fields}]}
+
+    async def push(client, payload: dict) -> dict:
+        return (await (await client.post("/api/sync/push", json=payload)).get_json())["results"][0]
+
+    viewed = await push(recipient, change(body="mine now"))
+    assert (viewed["status"], viewed["error"]) == ("rejected", "only its owner can change that")
+    probed = await push(stranger, change(body="mine now"))
+    assert (probed["status"], probed["error"]) == ("rejected", "cannot apply")
+
+    await _share(app_client, nid, people["recipient"], "edit")
+    edited = await push(recipient, change(body="first line, from the phone", pinned=True, archived=True, label_ids=[]))
+    assert edited["status"] == "applied", edited
+    note = await (await app_client.get(f"/api/notes/{nid}")).get_json()
+    assert note["body"] == "first line, from the phone"
+    # Everything but the text stays the owner's.
+    assert (note["pinned"], note["archived"]) == (False, False)
+
+    deleted = await push(recipient, {"changes": [{"entity": "note", "id": nid, "op": "delete", "edited_at": "2099-01-02T00:00:00Z"}]})
+    assert deleted["status"] == "rejected"
+    assert (await app_client.get(f"/api/notes/{nid}")).status_code == 200
+
+
+async def test_deleting_a_shared_note_reaches_the_recipients_devices(app_client, db):
+    recipient, _, people = await _three_people(app_client)
+    nid = await _owners_note(app_client)
+    await _share(app_client, nid, people["recipient"])
+    seen = await _feed(recipient)
+    assert _feed_note(seen, nid) is not None
+
+    assert (await app_client.post(f"/api/notes/{nid}/trash")).status_code == 200
+    trashed = _feed_note(await _feed(recipient, seen["cursor"]), nid)
+    assert trashed is not None and trashed["trashed"] is True
+
+    assert (await app_client.delete(f"/api/notes/{nid}")).status_code == 200
+    gone = await _feed(recipient, seen["cursor"])
+    assert gone["revoked"] == [nid]
+    assert _feed_note(gone, nid) is None
+
+
 # --- Password reset by email (#5266) ------------------------------------------
 #
 # The SMTP hand-off is replaced by a list; everything up to it is real.