sync

sync

Secure, local-first replication for the native Succès workspace.

The transport is deliberately separate from the data model: the Mac keeps working offline, every mutation remains in the immutable operation log, and a paired client exchanges batches using cursors. Pairing and peer credentials are stored only as SHA-256 hashes.

Classes

SuccesSyncStore

SuccesSyncStore(db_path: str | Path | None = None)

Bases: SuccesPhotosStore

Continuity store extended with authenticated operation replication.

Source code in src/diapason/succes/sync.py

def __init__(self, db_path: str | Path | None = None) -> None:
    super().__init__(db_path)
    with self._connect() as conn:
        conn.executescript(_SYNC_SCHEMA)
        self._ensure_peer_columns(conn)
        conn.execute(
            "INSERT OR IGNORE INTO succes_meta(key,value) "
            "VALUES('sync_clock_cursor','0')"
        )
        self._refresh_local_clocks(conn)
        conn.commit()

Methods:

apply_inbound_operations

apply_inbound_operations(
    operations: Sequence[Mapping[str, Any]],
) -> dict[str, int]

Apply a trusted exchange bundle locally (guest side, no peer row).

Source code in src/diapason/succes/sync.py

def apply_inbound_operations(
    self, operations: Sequence[Mapping[str, Any]]
) -> dict[str, int]:
    """Apply a trusted exchange bundle locally (guest side, no peer row)."""
    if len(operations) > MAX_SYNC_BATCH:
        raise SuccesError("Le lot reçu est trop volumineux.")
    stats = {"applied": 0, "stale": 0, "duplicate": 0}
    ordered = sorted(
        operations, key=lambda op: str(op.get("entity")) == "habit_logs"
    )
    with self._transaction() as conn:
        self._refresh_local_clocks(conn)
        for raw in ordered:
            result = self._apply_operation(conn, raw)
            stats[result] += 1
    return stats

join_remote

join_remote(
    pairing_token: str,
    *,
    relay_url: str | None = None,
    device_name: str = "",
) -> dict[str, Any]

Redeem a host invitation through the configured (or provided) relay.

Source code in src/diapason/succes/sync.py

def join_remote(
    self,
    pairing_token: str,
    *,
    relay_url: str | None = None,
    device_name: str = "",
) -> dict[str, Any]:
    """Redeem a host invitation through the configured (or provided) relay."""
    token = _clean_text(
        pairing_token,
        field="Le code d'appairage",
        maximum=160,
        required=True,
    )
    if not token.startswith("diapason_pair_"):
        raise SuccesError("Ce code d'appairage n'a pas un format reconnu.")
    base = normalize_relay_url(relay_url or self.relay_url())
    remote = relay_post(base, "/v1/succes/sync/pair", {"pairingToken": token})
    sync_token = str(remote.get("syncToken") or "")
    peer_id = str(remote.get("peerId") or "")
    if not sync_token.startswith("diapason_sync_") or not peer_id:
        raise SuccesError(
            "Le relais n'a pas renvoyé d'identifiants de sync valides."
        )
    name = _clean_text(
        device_name or remote.get("deviceName") or "Appareil distant",
        field="Le nom de l'appareil",
        maximum=80,
        required=True,
    )
    with self._transaction() as conn:
        self._meta_set(conn, _META_RELAY_URL, base)
        self._meta_set(conn, _META_GUEST_TOKEN, sync_token)
        self._meta_set(conn, _META_GUEST_PEER_ID, peer_id)
        self._meta_set(
            conn, _META_GUEST_SERVER_ID, str(remote.get("serverDeviceId") or "")
        )
        self._meta_set(conn, _META_GUEST_NAME, name)
        self._meta_set(conn, _META_GUEST_PULL, "0")
        self._meta_set(conn, _META_GUEST_PUSH, "0")
        self._meta_delete(conn, _META_LAST_SYNC_ERROR)
    status = self.sync_status()
    status["joined"] = {
        "peerId": peer_id,
        "deviceName": name,
        "serverDeviceId": remote.get("serverDeviceId"),
        "relayUrl": base,
    }
    return status

run_exchange

run_exchange() -> dict[str, Any]

Push local ops and pull host ops through the configured relay (guest).

Source code in src/diapason/succes/sync.py

def run_exchange(self) -> dict[str, Any]:
    """Push local ops and pull host ops through the configured relay (guest)."""
    base = self.relay_url()
    token = self._meta_get(_META_GUEST_TOKEN)
    if not base or not token:
        raise SuccesError(
            "Configurez d'abord un relais et rejoignez un appareil avec un code."
        )
    pull_cursor = int(self._meta_get(_META_GUEST_PULL) or 0)
    push_cursor = int(self._meta_get(_META_GUEST_PUSH) or 0)
    bundle = self.list_operations(after=push_cursor, limit=MAX_SYNC_BATCH)
    outbound = [
        {
            "opId": op["opId"],
            "deviceId": op["deviceId"],
            "entity": op["entity"],
            "entityId": op["entityId"],
            "kind": op["kind"],
            "request": op["request"],
            "payload": op["payload"],
            "timestampMs": op["timestampMs"],
        }
        for op in bundle["operations"]
        if op["deviceId"] == self.device_id()
    ]
    try:
        remote = relay_post(
            base,
            "/v1/succes/sync/exchange",
            {
                "peerToken": token,
                "cursor": pull_cursor,
                "operations": outbound,
            },
        )
    except SuccesError as exc:
        with self._transaction() as conn:
            self._meta_set(conn, _META_LAST_SYNC_ERROR, str(exc))
        raise

    inbound = remote.get("operations")
    if not isinstance(inbound, list):
        inbound = []
    received = self.apply_inbound_operations(inbound)
    next_pull = int(remote.get("cursor") or pull_cursor)
    if bundle["operations"]:
        next_push = int(bundle["operations"][-1]["cursor"])
    else:
        next_push = int(bundle["cursor"])
    timestamp = now_ms()
    with self._transaction() as conn:
        self._meta_set(conn, _META_GUEST_PULL, str(next_pull))
        self._meta_set(conn, _META_GUEST_PUSH, str(next_push))
        self._meta_set(conn, _META_LAST_SYNC_AT, str(timestamp))
        self._meta_delete(conn, _META_LAST_SYNC_ERROR)
    return {
        "pushed": len(outbound),
        "pulled": len(inbound),
        "received": received,
        "pullCursor": next_pull,
        "pushCursor": next_push,
        "hasMore": bool(remote.get("hasMore")) or bool(bundle.get("hasMore")),
        "syncedAtMs": timestamp,
        "status": self.sync_status(),
    }

Functions: