test_sync.py

# SPDX-License-Identifier: AGPL-3.0-or-later
# Copyright 2026 Chelsea Kelly-Reif
"""Peer-to-peer sync over a directory and over the relay."""

from __future__ import annotations

import base64
import os
import threading
import urllib.request
from collections.abc import Callable, Iterator
from pathlib import Path

import piexif
import pytest
from PIL import Image

from habitable import relay as relay_mod
from habitable.capture import capture, resolve_deferred
from habitable.errors import SyncError
from habitable.relay import RelayStore, make_server
from habitable.sync import (
    LocalDirTransport,
    PaddingTransport,
    RelayClient,
    export_message,
    import_messages,
    sync,
)
from habitable.tsa import LocalRfc3161TSA
from habitable.vault import Vault

SENTINEL = "PLAINTEXT-SENTINEL-mold-on-bathroom-ceiling"


def _seed(vault: Vault, make_jpeg: Callable[..., Path], tsa: LocalRfc3161TSA) -> str:
    issue = vault.document.add_issue(category="mold", room="bath", title=SENTINEL, issue_id="i1")
    capture(vault, make_jpeg(with_location=True), issue_id=issue, tsa=tsa)
    return issue


def test_directory_sync_converges(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    _seed(a, make_jpeg, local_tsa)
    transport = LocalDirTransport(tmp_path / "mbox")
    sync(a, b.identity.public(), transport, channel="room")
    result = sync(b, a.identity.public(), transport, channel="room")

    assert result.captures_imported == 1
    assert [i.issue_id for i in b.document.issues()] == ["i1"]
    capture_record = b.document.captures()[0]
    assert b.read_original(capture_record.capture_id, capture_record.content_hash)
    assert b.get_token(capture_record.capture_id) is not None
    assert b.custody.verify().ok

    # Idempotent: re-importing changes nothing.
    again = import_messages(b, transport.fetch("room"))
    assert again.captures_imported == 0


def _noisy_jpeg(path: Path, *, size: tuple[int, int] = (400, 300)) -> Path:
    """A larger, incompressible synthetic JPEG.

    ``make_jpeg`` makes a tiny 16x16 solid-color image (a few hundred bytes) — fine
    for most tests, but too small to show a size *ratio* effect: CRDT-state and
    envelope overhead would dwarf it. This is closer to a real phone photo, where
    the original dominates total message size, which is the regime FIX-02 targets.
    """
    width, height = size
    image = Image.frombytes("RGB", (width, height), os.urandom(width * height * 3))
    exif = {piexif.ExifIFD.DateTimeOriginal: b"2026:01:02 03:04:05"}
    gps = {
        piexif.GPSIFD.GPSLatitudeRef: b"N",
        piexif.GPSIFD.GPSLatitude: ((38, 1), (33, 1), (0, 1)),
        piexif.GPSIFD.GPSLongitudeRef: b"W",
        piexif.GPSIFD.GPSLongitude: ((121, 1), (44, 1), (0, 1)),
    }
    payload = {"0th": {}, "Exif": exif, "GPS": gps, "1st": {}, "thumbnail": None}
    image.save(path, "jpeg", exif=piexif.dump(payload))
    return path


def test_incremental_sync_skips_already_held_originals(
    make_vault: Callable[..., Vault],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    """FIX-02: once a peer has confirmed holding a capture, re-sending it is skipped.

    A full round trip (A->B, B->A) is needed before B's inventory is confirmed back
    to A, since neither side knows the other's holdings until told. After that
    steady state: re-syncing an unchanged case carries ~no original bytes, and
    adding one more capture only costs that one capture, not the whole case.
    """
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    for n in range(5):
        issue = a.document.add_issue(
            category="mold", room=f"room-{n}", title=SENTINEL, issue_id=f"i{n}"
        )
        capture(a, _noisy_jpeg(tmp_path / f"photo-{n}.jpg"), issue_id=issue, tsa=local_tsa)
    transport = LocalDirTransport(tmp_path / "mbox")

    # Round 1: full exchange establishes mutual inventory.
    first_export = export_message(a, b.identity.public())
    sync(a, b.identity.public(), transport, channel="room")  # A -> B: full send (B has nothing)
    sync(b, a.identity.public(), transport, channel="room")  # B imports A's captures
    # A second B->A leg: B now knows (from A's "have") that A has everything, so this
    # leg tells A nothing new about its own captures but does carry B's "have" back.
    sync(b, a.identity.public(), transport, channel="room")
    # A reads the channel to pick up B's declared "have" manifests.
    import_messages(a, transport.fetch("room"))
    assert b.document.captures() and len(b.document.captures()) == 5
    assert b.custody.verify().ok
    assert a.known_peer_captures(b.identity.public().fingerprint) == {
        c.capture_id for c in a.document.captures()
    }

    # Round 2: nothing new — A's export to B should carry ~none of the original bytes.
    steady_export = export_message(a, b.identity.public())
    assert len(steady_export) < 0.05 * len(first_export)

    # Round 3: one new capture — the export grows by about one photo, not the whole case.
    issue = a.document.add_issue(category="mold", room="room-new", title=SENTINEL, issue_id="i-new")
    capture(a, _noisy_jpeg(tmp_path / "photo-new.jpg"), issue_id=issue, tsa=local_tsa)
    delta_export = export_message(a, b.identity.public())
    per_photo = (len(first_export) - len(steady_export)) / 5
    assert len(delta_export) - len(steady_export) < 2 * per_photo


def test_message_isolation(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    c = make_vault("C", passphrase="pw-c")
    _seed(a, make_jpeg, local_tsa)
    transport = LocalDirTransport(tmp_path / "mbox")
    sync(a, b.identity.public(), transport, channel="room")  # sealed to B only
    # C cannot open a message addressed to B.
    result = import_messages(c, transport.fetch("room"))
    assert result.captures_imported == 0 and result.messages_merged == 0


@pytest.fixture
def relay_url() -> Iterator[tuple[str, RelayStore]]:
    store = RelayStore()
    server = make_server("127.0.0.1", 0, store)
    port = server.server_address[1]
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield f"http://127.0.0.1:{port}", store
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def _no_content_markers(a: Vault, image_bytes: bytes) -> list[bytes]:
    """Every plaintext that must NEVER appear in a stored/transmitted blob."""
    return [
        SENTINEL.encode(),  # the note text (rides in the CRDT state)
        image_bytes[:64],  # raw original image bytes (ride in captures[].original_b64)
        base64.b64encode(image_bytes)[:64],  # ...nor their base64 form
        a.identity.public().fingerprint.encode(),  # the sender identity (in the envelope)
    ]


def test_relay_sync_is_end_to_end_encrypted(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    relay_url: tuple[str, RelayStore],
) -> None:
    url, store = relay_url
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    issue = a.document.add_issue(category="mold", room="bath", title=SENTINEL, issue_id="i1")
    photo = make_jpeg("unique-source.jpg", with_location=True)
    image_bytes = photo.read_bytes()
    capture(a, photo, issue_id=issue, tsa=local_tsa)

    client = RelayClient(url)
    sync(a, b.identity.public(), client, channel="room-relay")
    result = sync(b, a.identity.public(), client, channel="room-relay")

    assert result.captures_imported == 1
    assert b.custody.verify().ok
    # "Ciphertext in, ciphertext out": no plaintext of any kind — note text, raw or
    # base64 image bytes, or the sender's own identity — appears in a stored blob or in
    # the base64 the relay GET handler serves back.
    blobs = store.fetch("room-relay")
    assert blobs
    served_back = b"".join(base64.b64encode(blob) for blob in blobs)
    for marker in _no_content_markers(a, image_bytes):
        for blob in blobs:
            assert marker not in blob
        assert marker not in served_back
    posted = store.metrics()["posted"]  # metrics carries counters and replay-state strings
    assert isinstance(posted, int) and posted >= 2


def test_relay_client_sends_a_matching_room_token(
    relay_url: tuple[str, RelayStore],
) -> None:
    """Both peers derive the same per-channel token, so writes round-trip (no 403)."""
    url, store = relay_url
    client = RelayClient(url)
    client.post("room-token", b"sealed-1")
    client.post("room-token", b"sealed-2")  # same channel -> same token -> accepted
    assert store.fetch("room-token") == [b"sealed-1", b"sealed-2"]


def test_relay_client_raises_clear_error_when_room_full(
    relay_url: tuple[str, RelayStore],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    url, _store = relay_url
    monkeypatch.setattr(relay_mod, "_MAX_MESSAGES_PER_ROOM", 1)
    client = RelayClient(url)
    client.post("room-full", b"first")  # fills the room
    with pytest.raises(SyncError, match="full") as raised:
        client.post("room-full", b"second")
    message = str(raised.value)
    assert "GET does not clear" in message
    assert "TTL and retry" in message
    assert "fetch and clear" not in message


def test_relay_client_raises_clear_error_when_token_rejected(
    relay_url: tuple[str, RelayStore],
) -> None:
    url, store = relay_url
    # Someone else claims the room first with a different token (trust-on-first-use).
    store.post("room-claimed", b"squatter", token="not-the-derived-token")
    client = RelayClient(url)
    with pytest.raises(SyncError, match="token"):
        client.post("room-claimed", b"mine")


@pytest.mark.parametrize(
    ("body", "message"),
    [
        (b"not-json", "invalid JSON"),
        (b'{"messages":["***not-base64***"]}', "invalid base64"),
        (b'{"messages":[42]}', "not base64 text"),
    ],
)
def test_relay_client_rejects_malformed_fetch_responses(
    monkeypatch: pytest.MonkeyPatch, body: bytes, message: str
) -> None:
    class Response:
        def __enter__(self) -> Response:
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def read(self) -> bytes:
            return body

    monkeypatch.setattr(urllib.request, "urlopen", lambda *_a, **_kw: Response())
    with pytest.raises(SyncError, match=message):
        RelayClient("https://relay.example").fetch("room")


def test_localdir_mailbox_holds_only_ciphertext(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    issue = a.document.add_issue(category="mold", room="bath", title=SENTINEL, issue_id="i1")
    photo = make_jpeg("unique-source.jpg", with_location=True)
    image_bytes = photo.read_bytes()
    capture(a, photo, issue_id=issue, tsa=local_tsa)

    mbox_dir = tmp_path / "mbox"
    transport = LocalDirTransport(mbox_dir)
    sync(a, b.identity.public(), transport, channel="room")

    # The on-disk mailbox holds only base64 of sealed blobs: assert no plaintext marker
    # survives in either the raw file bytes or any base64-decoded line.
    raw = b"".join(p.read_bytes() for p in mbox_dir.glob("*"))
    decoded = b"".join(base64.b64decode(line) for line in raw.splitlines() if line.strip())
    assert raw and decoded
    for marker in _no_content_markers(a, image_bytes):
        assert marker not in raw
        assert marker not in decoded


# --- metadata-resistant transport (EXP-12) ------------------------------------


def test_padding_transport_round_trips_a_full_sync(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    """Wrapping a transport in padding + cover traffic must not break real delivery."""
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    _seed(a, make_jpeg, local_tsa)
    inner = LocalDirTransport(tmp_path / "mbox")
    ta = PaddingTransport(inner, block_size=4096, batch_size=4)
    tb = PaddingTransport(inner, block_size=4096, batch_size=4)
    sync(a, b.identity.public(), ta, channel="room")
    result = sync(b, a.identity.public(), tb, channel="room")

    # The real message survives the padding/decoy round-trip; decoys are dropped silently.
    assert result.captures_imported == 1
    assert [i.issue_id for i in b.document.issues()] == ["i1"]
    assert b.custody.verify().ok

    # Idempotent even through padding: re-importing changes nothing.
    again = import_messages(b, tb.fetch("room"))
    assert again.captures_imported == 0


def test_padding_transport_emits_uniform_block_sized_cover_batch(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    """One real post must leave the relay a full batch of identical-size blobs."""
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    _seed(a, make_jpeg, local_tsa)
    inner = LocalDirTransport(tmp_path / "mbox")
    transport = PaddingTransport(inner, block_size=4096, batch_size=4)
    sync(a, b.identity.public(), transport, channel="room")

    # What the relay actually stored: read the raw framed blobs via the inner transport.
    raw_blobs = inner.fetch("room")
    # Exactly batch_size blobs left the sender (1 real + 3 decoys) — the relay cannot tell
    # from the count how many were real.
    assert len(raw_blobs) == 4
    # Every blob in the flush is padded to one identical, block-aligned size, so neither
    # size nor position distinguishes the real message from its decoys.
    sizes = {len(blob) for blob in raw_blobs}
    assert len(sizes) == 1
    assert next(iter(sizes)) % 4096 == 0


def test_padding_transport_drops_decoys_on_import(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    """A channel of mostly decoys imports exactly the one real message, no more."""
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    _seed(a, make_jpeg, local_tsa)
    inner = LocalDirTransport(tmp_path / "mbox")
    # A large batch means many decoys accompany the single real message.
    transport = PaddingTransport(inner, block_size=4096, batch_size=8)
    sync(a, b.identity.public(), transport, channel="room")
    assert len(inner.fetch("room")) == 8  # 1 real + 7 decoys on the wire

    tb = PaddingTransport(inner, block_size=4096, batch_size=8)
    result = import_messages(b, tb.fetch("room"))
    assert result.messages_merged == 1
    assert result.captures_imported == 1


def test_padding_transport_batches_multiple_posts_when_not_auto_flushing(
    make_vault: Callable[..., Vault],
    tmp_path: Path,
) -> None:
    """auto_flush=False buffers posts until one flush emits a single padded batch."""
    inner = LocalDirTransport(tmp_path / "mbox")
    transport = PaddingTransport(inner, block_size=1024, batch_size=4, auto_flush=False)
    transport.post("room", b"one")
    transport.post("room", b"two")
    # Nothing has left the sender yet: buffered, not posted.
    assert inner.fetch("room") == []

    transport.flush("room")
    raw_blobs = inner.fetch("room")
    # Two real + two decoys, all one block, emitted together in a single batch.
    assert len(raw_blobs) == 4
    assert {len(blob) for blob in raw_blobs} == {1024}
    # Both real payloads are recoverable (order is shuffled, so compare as a set).
    recovered = {transport._unframe(blob) for blob in raw_blobs}
    assert {b"one", b"two"} <= recovered


def test_padding_transport_passes_through_unframed_blobs(tmp_path: Path) -> None:
    """A channel that also carries plain (unpadded) blobs still delivers them."""
    inner = LocalDirTransport(tmp_path / "mbox")
    inner.post("room", b"legacy-unframed-message")
    transport = PaddingTransport(inner, block_size=1024, batch_size=2)
    assert b"legacy-unframed-message" in transport.fetch("room")


def test_an_imported_capture_with_no_token_is_queued_so_resolve_can_reach_it(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    """Issue #180: sync wrote nothing to the deferred queue, so `resolve` could not
    reach a capture that arrived without a token -- it was untimestamped forever."""
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    issue = a.document.add_issue(category="no_heat", title="No heat", issue_id="i1")
    capture(a, make_jpeg(), issue_id=issue, tsa=None)  # offline: no token

    transport = LocalDirTransport(tmp_path / "mbox")
    sync(a, b.identity.public(), transport, channel="room")
    sync(b, a.identity.public(), transport, channel="room")

    capture_id = b.document.captures()[0].capture_id
    assert b.get_token(capture_id) is None
    assert b.awaiting_timestamp() == (capture_id,)
    assert [item.capture_id for item in b.deferred()] == [capture_id]

    assert len(resolve_deferred(b, local_tsa)) == 1
    assert b.get_token(capture_id) is not None
    assert b.awaiting_timestamp() == ()


def test_reimporting_the_same_untimestamped_capture_does_not_double_queue_it(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    tmp_path: Path,
) -> None:
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    issue = a.document.add_issue(category="no_heat", title="No heat", issue_id="i1")
    capture(a, make_jpeg(), issue_id=issue, tsa=None)

    transport = LocalDirTransport(tmp_path / "mbox")
    for _ in range(2):
        sync(a, b.identity.public(), transport, channel="room")
        sync(b, a.identity.public(), transport, channel="room")

    assert len(b.deferred()) == 1


def test_a_peers_token_clears_our_own_queued_entry_for_that_capture(
    make_vault: Callable[..., Vault],
    make_jpeg: Callable[..., Path],
    local_tsa: LocalRfc3161TSA,
    tmp_path: Path,
) -> None:
    """Otherwise `resolve` fetches a second primary over content already stamped."""
    a = make_vault("A")
    b = make_vault("B", passphrase="pw-b")
    issue = a.document.add_issue(category="no_heat", title="No heat", issue_id="i1")
    capture(a, make_jpeg(), issue_id=issue, tsa=None)

    transport = LocalDirTransport(tmp_path / "mbox")
    sync(a, b.identity.public(), transport, channel="room")
    sync(b, a.identity.public(), transport, channel="room")
    assert len(b.deferred()) == 1

    # A now gets its token and syncs again.
    assert len(resolve_deferred(a, local_tsa)) == 1
    sync(a, b.identity.public(), transport, channel="room")
    sync(b, a.identity.public(), transport, channel="room")

    capture_id = b.document.captures()[0].capture_id
    assert b.get_token(capture_id) is not None
    assert b.deferred() == ()
    assert b.awaiting_timestamp() == ()