kinqsradio · Nov 11, 2026 · Nov 11, 2026 · Nov 11, 2026 · Nov 11, 2026
@@ -10,14 +10,18 @@ publish.workspace = true
 [dependencies]
 base64.workspace = true
 getrandom.workspace = true
+futures-util = { version = "0.3", default-features = false, features = ["sink"] }
 orbyn-auth-client.workspace = true
+orbyn-client.workspace = true
 orbyn-contracts.workspace = true
 orbyn-crypto.workspace = true
+orbyn-sync.workspace = true
 reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
+tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] }
 uuid.workspace = true
 zeroize.workspace = true
 

@@ -56,6 +56,12 @@
 /// One call to the engine. A failure to get any answer is [`DeviceError::Transport`].
 pub trait DeviceApi: Send + Sync {
     fn send(&self, request: Request) -> impl Future<Output = Result<Response, DeviceError>> + Send;
+
+    /// The engine's address, for the live notices stream (a WebSocket). A transport that is not a
+    /// network (a test double) has none, and then there is no notices stream.
+    fn base_url(&self) -> Option<&str> {
+        None
+    }
 }
 
 /// Whether `address` is a URL whose host is a loopback **IP literal** (`127.0.0.1`, `[::1]`).
@@ -127,13 +133,17 @@
 }
 
 impl DeviceApi for HttpApi {
+    fn base_url(&self) -> Option<&str> {
+        Some(&self.base)
+    }
+
     async fn send(&self, request: Request) -> Result<Response, DeviceError> {
         let url = format!("{}{}", self.base, request.path);
         let mut builder = match request.method {
            Method::Get => self.client.get(url),
            Method::Post => self.client.post(url),
            Method::Put => self.client.put(url),
            Method::Delete => self.client.delete(url),
        };
        if let Some(token) = &request.bearer {
            builder = builder.bearer_auth(token.as_str());

@@ -4,6 +4,7 @@
 use crate::api::{DeviceApi, Method, Request, Response};
 use crate::clock::Clock;
 use crate::error::DeviceError;
+use crate::status::SyncStatus;
 use crate::store::{KeyStore, StoredIdentity, StoredSession};
 use base64::Engine as _;
 use base64::engine::general_purpose::URL_SAFE_NO_PAD as B64;
@@ -13,6 +14,7 @@ use orbyn_crypto::{DeviceKeys, DevicePublic};
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use std::time::Duration;
+use tokio::sync::watch;
 use zeroize::Zeroizing;
 
 /// A session as the server answers it. The token is wiped when this is dropped.
@@ -102,6 +104,7 @@ pub struct DeviceClient<A, S, C> {
     pub(crate) store: S,
     pub(crate) clock: C,
     pub(crate) config: ClientConfig,
+    pub(crate) status: watch::Sender<SyncStatus>,
 }
 
 pub(crate) fn encode(bytes: &[u8]) -> String {
@@ -170,9 +173,15 @@ impl<A: DeviceApi, S: KeyStore, C: Clock> DeviceClient<A, S, C> {
             store,
             clock,
             config,
+            status: watch::Sender::new(SyncStatus::initial()),
         }
     }
 
+    /// The transport, for a host that wraps it or reads from it.
+    pub fn api(&self) -> &A {
+        &self.api
+    }
+
     /// The key store, for the host.
     pub fn store(&self) -> &S {
         &self.store
@@ -227,10 +236,11 @@ impl<A: DeviceApi, S: KeyStore, C: Clock> DeviceClient<A, S, C> {
         parse(&response.body)
     }
 
-    /// An authenticated call with this device's session. A 401 asks the server whether the
-    /// account still exists: [`DeviceError::AccountGone`] if it was deleted, else
+    /// An authenticated call with this device's session, whatever the status: the caller reads
+    /// the answer. A 401 `unauthorized` is asked about: [`DeviceError::AccountGone`] only if
+    /// `POST /v1/accounts/status` answers 200 with a deletion time, else
     /// [`DeviceError::SessionExpired`].
-    pub(crate) async fn authed(
+    pub(crate) async fn authed_raw(
         &self,
         method: Method,
         path: &str,
@@ -241,11 +251,7 @@ impl<A: DeviceApi, S: KeyStore, C: Clock> DeviceClient<A, S, C> {
         let response = self
             .call(method, path, Some(session.token.clone()), body)
             .await?;
-        if is_ok(&response) {
-            return Ok(response);
-        }
-        let error = failure(&response);
-        if error == DeviceError::SessionExpired {
+        if failure(&response) == DeviceError::SessionExpired {
             let status: Result<wire::AccountStatus, _> = self
                 .post(
                     "/v1/accounts/status",
@@ -257,8 +263,24 @@ impl<A: DeviceApi, S: KeyStore, C: Clock> DeviceClient<A, S, C> {
             if status.is_ok_and(|s| !s.deleted_at.is_empty()) {
                 return Err(DeviceError::AccountGone);
             }
+            return Err(DeviceError::SessionExpired);
+        }
+        Ok(response)
+    }
+
+    /// An authenticated call that must answer 2xx.
+    pub(crate) async fn authed(
+        &self,
+        method: Method,
+        path: &str,
+        body: Option<Zeroizing<Vec<u8>>>,
+    ) -> Result<Response, DeviceError> {
+        let response = self.authed_raw(method, path, body).await?;
+        if is_ok(&response) {
+            Ok(response)
+        } else {
+            Err(failure(&response))
         }
-        Err(error)
     }
 
     // ---- sign-in ----

@@ -56,6 +56,16 @@ pub enum DeviceError {
     /// A value the caller gave is not valid (an email, an address, a name).
     #[error("{0}")]
     InvalidInput(&'static str),
+    /// The zone was deleted, here or on another device.
+    #[error("the zone was deleted")]
+    ZoneDeleted,
+    /// This device has no key for the zone yet: another device sends it.
+    #[error("this device has no key for the zone yet")]
+    NoZoneKey,
+    /// The sync engine refused something (a record, a key, a snapshot). The message never holds
+    /// record content or keys.
+    #[error("sync engine: {0}")]
+    Engine(String),
     /// The server answered something this client does not handle.
     #[error("the server answered {status} {code}")]
     Server { status: u16, code: String },

@@ -7,8 +7,11 @@ mod api;
 mod client;
 mod clock;
 mod error;
+mod notices;
 mod signup;
+mod status;
 mod store;
+mod sync;
 
 pub use api::{DeviceApi, HttpApi, Method, Request, Response, is_loopback_literal};
 pub use client::{
@@ -17,5 +20,8 @@ pub use client::{
 };
 pub use clock::{Clock, SystemClock};
 pub use error::{DeviceError, StoreError};
+pub use notices::{Notice, NoticeConfig, Notices};
 pub use signup::PendingSignUp;
+pub use status::{SyncState, SyncStatus};
 pub use store::{KeyStore, MemoryStore, StoredIdentity, StoredSession};
+pub use sync::{SyncReport, ZoneReport};