use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;

use anyhow::Context;
use quinn::{ClientConfig, Endpoint};
use quinn_proto::crypto::rustls::QuicClientConfig;
use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
use rustls::{ClientConfig as RustlsClientConfig, RootCertStore};
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
use capnp_rpc::{rpc_twoparty_capnp::Side, twoparty, RpcSystem};

use quicproquo_core::HybridPublicKey;
use quicproquo_proto::node_capnp::{auth, node_service};

use crate::{AUTH_CONTEXT, INSECURE_SKIP_VERIFY};

use super::retry::{anyhow_is_retriable, retry_async, DEFAULT_BASE_DELAY_MS, DEFAULT_MAX_RETRIES};

/// Cap'n Proto traversal limit (words). 4 Mi words = 32 MiB; bounds DoS from deeply nested or large messages.
const CAPNP_TRAVERSAL_LIMIT_WORDS: usize = 4 * 1024 * 1024;

/// A [`rustls::client::danger::ServerCertVerifier`] that accepts any certificate.
///
/// **Development only.** Using this in production disables all TLS guarantees.
#[derive(Debug)]
struct InsecureServerCertVerifier;

impl rustls::client::danger::ServerCertVerifier for InsecureServerCertVerifier {
    fn verify_server_cert(
        &self,
        _end_entity: &CertificateDer<'_>,
        _intermediates: &[CertificateDer<'_>],
        _server_name: &ServerName<'_>,
        _ocsp_response: &[u8],
        _now: UnixTime,
    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
        Ok(rustls::client::danger::ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &CertificateDer<'_>,
        _dss: &rustls::DigitallySignedStruct,
    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &CertificateDer<'_>,
        _dss: &rustls::DigitallySignedStruct,
    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
        rustls::crypto::ring::default_provider()
            .signature_verification_algorithms
            .supported_schemes()
    }
}

/// Establish a QUIC/TLS connection and return a `NodeService` client.
///
/// Must be called from within a `LocalSet` because capnp-rpc is `!Send`.
///
/// Reads [`INSECURE_SKIP_VERIFY`] to decide whether to bypass certificate
/// verification (set once at startup via [`crate::set_insecure_skip_verify`]).
pub async fn connect_node(
    server: &str,
    ca_cert: &Path,
    server_name: &str,
) -> anyhow::Result<node_service::Client> {
    let skip = INSECURE_SKIP_VERIFY.load(std::sync::atomic::Ordering::Relaxed);
    connect_node_opt(server, ca_cert, server_name, skip).await
}

/// Like [`connect_node`] but with an explicit `insecure_skip_verify` toggle.
///
/// When `insecure_skip_verify` is `true`, certificate verification is disabled entirely.
/// This is intended for development and testing only.
pub async fn connect_node_opt(
    server: &str,
    ca_cert: &Path,
    server_name: &str,
    insecure_skip_verify: bool,
) -> anyhow::Result<node_service::Client> {
    let addr: SocketAddr = server
        .parse()
        .with_context(|| format!("server must be host:port, got {server}"))?;

    let mut tls = if insecure_skip_verify {
        RustlsClientConfig::builder()
            .dangerous()
            .with_custom_certificate_verifier(Arc::new(InsecureServerCertVerifier))
            .with_no_client_auth()
    } else {
        let cert_bytes =
            std::fs::read(ca_cert).with_context(|| format!("read ca_cert {ca_cert:?}"))?;
        let mut roots = RootCertStore::empty();
        roots
            .add(CertificateDer::from(cert_bytes))
            .context("add root cert")?;
        RustlsClientConfig::builder()
            .with_root_certificates(roots)
            .with_no_client_auth()
    };
    tls.alpn_protocols = vec![b"capnp".to_vec()];

    let crypto = QuicClientConfig::try_from(tls)
        .map_err(|e| anyhow::anyhow!("invalid client TLS config: {e}"))?;

    let bind_addr: SocketAddr = "0.0.0.0:0".parse().context("parse client bind address")?;
    let mut endpoint = Endpoint::client(bind_addr)?;
    endpoint.set_default_client_config(ClientConfig::new(Arc::new(crypto)));

    let connection = endpoint
        .connect(addr, server_name)
        .context("quic connect init")?
        .await
        .context("quic connect failed")?;

    let (send, recv) = connection.open_bi().await.context("open bi stream")?;

    let mut reader_opts = capnp::message::ReaderOptions::new();
    reader_opts.traversal_limit_in_words(Some(CAPNP_TRAVERSAL_LIMIT_WORDS));
    let network = twoparty::VatNetwork::new(
        recv.compat(),
        send.compat_write(),
        Side::Client,
        reader_opts,
    );

    let mut rpc_system = RpcSystem::new(Box::new(network), None);
    let client: node_service::Client = rpc_system.bootstrap(Side::Server);

    tokio::task::spawn_local(rpc_system);

    Ok(client)
}

pub fn set_auth(auth: &mut auth::Builder<'_>) -> anyhow::Result<()> {
    let guard = AUTH_CONTEXT
        .read()
        .map_err(|e| anyhow::anyhow!("AUTH_CONTEXT lock poisoned: {e}"))?;
    let ctx = guard.as_ref().ok_or_else(|| {
        anyhow::anyhow!(
            "init_auth must be called before RPCs (use a bearer or session token for authenticated commands)"
        )
    })?;
    auth.set_version(ctx.version);
    auth.set_access_token(&ctx.access_token);
    auth.set_device_id(&ctx.device_id);
    Ok(())
}

/// Upload a KeyPackage and verify the fingerprint echoed by the AS.
pub async fn upload_key_package(
    client: &node_service::Client,
    identity_key: &[u8],
    package: &[u8],
) -> anyhow::Result<()> {
    let mut req = client.upload_key_package_request();
    {
        let mut p = req.get();
        p.set_identity_key(identity_key);
        p.set_package(package);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("upload_key_package RPC failed")?;

    let server_fp = resp
        .get()
        .context("upload_key_package: bad response")?
        .get_fingerprint()
        .context("upload_key_package: missing fingerprint")?
        .to_vec();

    let local_fp = super::state::sha256(package);
    anyhow::ensure!(server_fp == local_fp, "fingerprint mismatch");
    Ok(())
}

/// Fetch a KeyPackage for `identity_key` from the AS.
pub async fn fetch_key_package(
    client: &node_service::Client,
    identity_key: &[u8],
) -> anyhow::Result<Vec<u8>> {
    let mut req = client.fetch_key_package_request();
    {
        let mut p = req.get();
        p.set_identity_key(identity_key);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("fetch_key_package RPC failed")?;

    let pkg = resp
        .get()
        .context("fetch_key_package: bad response")?
        .get_package()
        .context("fetch_key_package: missing package field")?
        .to_vec();

    Ok(pkg)
}

/// Enqueue an opaque payload to the DS for `recipient_key`.
/// Returns the per-inbox sequence number assigned by the server.
/// Retries on transient failures with exponential backoff.
pub async fn enqueue(
    client: &node_service::Client,
    recipient_key: &[u8],
    payload: &[u8],
) -> anyhow::Result<u64> {
    enqueue_with_ttl(client, recipient_key, payload, None).await
}

/// Enqueue with an optional TTL (seconds). 0 or None means no expiry.
pub async fn enqueue_with_ttl(
    client: &node_service::Client,
    recipient_key: &[u8],
    payload: &[u8],
    ttl_secs: Option<u32>,
) -> anyhow::Result<u64> {
    let client = client.clone();
    let recipient_key = recipient_key.to_vec();
    let payload = payload.to_vec();
    retry_async(
        || {
            let client = client.clone();
            let recipient_key = recipient_key.clone();
            let payload = payload.clone();
            async move {
                let mut req = client.enqueue_request();
                {
                    let mut p = req.get();
                    p.set_recipient_key(&recipient_key);
                    p.set_payload(&payload);
                    p.set_channel_id(&[]);
                    p.set_version(1);
                    if let Some(ttl) = ttl_secs {
                        p.set_ttl_secs(ttl);
                    }
                    let mut auth = p.reborrow().init_auth();
                    set_auth(&mut auth)?;
                }
                let resp = req.send().promise.await.context("enqueue RPC failed")?;
                let seq = resp.get().context("enqueue: bad response")?.get_seq();
                Ok(seq)
            }
        },
        DEFAULT_MAX_RETRIES,
        DEFAULT_BASE_DELAY_MS,
        anyhow_is_retriable,
    )
    .await
}

/// Fetch and drain all payloads for `recipient_key`.
/// Returns `(seq, payload)` pairs — sort by `seq` before MLS processing.
/// Retries on transient failures with exponential backoff.
pub async fn fetch_all(
    client: &node_service::Client,
    recipient_key: &[u8],
) -> anyhow::Result<Vec<(u64, Vec<u8>)>> {
    let client = client.clone();
    let recipient_key = recipient_key.to_vec();
    retry_async(
        || {
            let client = client.clone();
            let recipient_key = recipient_key.clone();
            async move {
                let mut req = client.fetch_request();
                {
                    let mut p = req.get();
                    p.set_recipient_key(&recipient_key);
                    p.set_channel_id(&[]);
                    p.set_version(1);
                    p.set_limit(0); // fetch all
                    let mut auth = p.reborrow().init_auth();
                    set_auth(&mut auth)?;
                }

                let resp = req.send().promise.await.context("fetch RPC failed")?;

                let list = resp
                    .get()
                    .context("fetch: bad response")?
                    .get_payloads()
                    .context("fetch: missing payloads")?;

                let mut payloads = Vec::with_capacity(list.len() as usize);
                for i in 0..list.len() {
                    let entry = list.get(i);
                    let seq = entry.get_seq();
                    let data = entry
                        .get_data()
                        .context("fetch: envelope data read failed")?
                        .to_vec();
                    payloads.push((seq, data));
                }

                Ok(payloads)
            }
        },
        DEFAULT_MAX_RETRIES,
        DEFAULT_BASE_DELAY_MS,
        anyhow_is_retriable,
    )
    .await
}

/// Long-poll for payloads with optional timeout (ms).
/// Returns `(seq, payload)` pairs — sort by `seq` before MLS processing.
/// Retries on transient failures with exponential backoff.
pub async fn fetch_wait(
    client: &node_service::Client,
    recipient_key: &[u8],
    timeout_ms: u64,
) -> anyhow::Result<Vec<(u64, Vec<u8>)>> {
    let client = client.clone();
    let recipient_key = recipient_key.to_vec();
    retry_async(
        || {
            let client = client.clone();
            let recipient_key = recipient_key.clone();
            async move {
                let mut req = client.fetch_wait_request();
                {
                    let mut p = req.get();
                    p.set_recipient_key(&recipient_key);
                    p.set_timeout_ms(timeout_ms);
                    p.set_channel_id(&[]);
                    p.set_version(1);
                    p.set_limit(0); // fetch all
                    let mut auth = p.reborrow().init_auth();
                    set_auth(&mut auth)?;
                }

                let resp = req.send().promise.await.context("fetch_wait RPC failed")?;

                let list = resp
                    .get()
                    .context("fetch_wait: bad response")?
                    .get_payloads()
                    .context("fetch_wait: missing payloads")?;

                let mut payloads = Vec::with_capacity(list.len() as usize);
                for i in 0..list.len() {
                    let entry = list.get(i);
                    let seq = entry.get_seq();
                    let data = entry
                        .get_data()
                        .context("fetch_wait: envelope data read failed")?
                        .to_vec();
                    payloads.push((seq, data));
                }

                Ok(payloads)
            }
        },
        DEFAULT_MAX_RETRIES,
        DEFAULT_BASE_DELAY_MS,
        anyhow_is_retriable,
    )
    .await
}

/// Upload a hybrid (X25519 + ML-KEM-768) public key for an identity.
pub async fn upload_hybrid_key(
    client: &node_service::Client,
    identity_key: &[u8],
    hybrid_pk: &HybridPublicKey,
) -> anyhow::Result<()> {
    let mut req = client.upload_hybrid_key_request();
    {
        let mut p = req.get();
        p.set_identity_key(identity_key);
        p.set_hybrid_public_key(&hybrid_pk.to_bytes());
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }
    req.send()
        .promise
        .await
        .context("upload_hybrid_key RPC failed")?;
    Ok(())
}

/// Fetch a peer's hybrid public key from the server.
///
/// Returns `None` if the peer has not uploaded a hybrid key.
pub async fn fetch_hybrid_key(
    client: &node_service::Client,
    identity_key: &[u8],
) -> anyhow::Result<Option<HybridPublicKey>> {
    let mut req = client.fetch_hybrid_key_request();
    {
        let mut p = req.get();
        p.set_identity_key(identity_key);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("fetch_hybrid_key RPC failed")?;

    let pk_bytes = resp
        .get()
        .context("fetch_hybrid_key: bad response")?
        .get_hybrid_public_key()
        .context("fetch_hybrid_key: missing field")?
        .to_vec();

    if pk_bytes.is_empty() {
        return Ok(None);
    }

    let pk = HybridPublicKey::from_bytes(&pk_bytes).context("invalid hybrid public key")?;
    Ok(Some(pk))
}

/// Decrypt a hybrid envelope. Requires a hybrid key; no fallback to plaintext MLS.
pub fn try_hybrid_decrypt(
    hybrid_kp: Option<&quicproquo_core::HybridKeypair>,
    payload: &[u8],
) -> anyhow::Result<Vec<u8>> {
    let kp = hybrid_kp.ok_or_else(|| anyhow::anyhow!("hybrid key required for decryption"))?;
    quicproquo_core::hybrid_decrypt(kp, payload, b"", b"").map_err(|e| anyhow::anyhow!("{e}"))
}

/// Peek at queued payloads without removing them.
/// Returns `(seq, payload)` pairs sorted by seq.
/// Retries on transient failures with exponential backoff.
pub async fn peek(
    client: &node_service::Client,
    recipient_key: &[u8],
) -> anyhow::Result<Vec<(u64, Vec<u8>)>> {
    let client = client.clone();
    let recipient_key = recipient_key.to_vec();
    retry_async(
        || {
            let client = client.clone();
            let recipient_key = recipient_key.clone();
            async move {
                let mut req = client.peek_request();
                {
                    let mut p = req.get();
                    p.set_recipient_key(&recipient_key);
                    p.set_channel_id(&[]);
                    p.set_version(1);
                    p.set_limit(0); // peek all
                    let mut auth = p.reborrow().init_auth();
                    set_auth(&mut auth)?;
                }

                let resp = req.send().promise.await.context("peek RPC failed")?;

                let list = resp
                    .get()
                    .context("peek: bad response")?
                    .get_payloads()
                    .context("peek: missing payloads")?;

                let mut payloads = Vec::with_capacity(list.len() as usize);
                for i in 0..list.len() {
                    let entry = list.get(i);
                    let seq = entry.get_seq();
                    let data = entry
                        .get_data()
                        .context("peek: envelope data read failed")?
                        .to_vec();
                    payloads.push((seq, data));
                }

                Ok(payloads)
            }
        },
        DEFAULT_MAX_RETRIES,
        DEFAULT_BASE_DELAY_MS,
        anyhow_is_retriable,
    )
    .await
}

/// Acknowledge all messages up to and including `seq_up_to`.
/// Retries on transient failures with exponential backoff.
pub async fn ack(
    client: &node_service::Client,
    recipient_key: &[u8],
    seq_up_to: u64,
) -> anyhow::Result<()> {
    let client = client.clone();
    let recipient_key = recipient_key.to_vec();
    retry_async(
        || {
            let client = client.clone();
            let recipient_key = recipient_key.clone();
            async move {
                let mut req = client.ack_request();
                {
                    let mut p = req.get();
                    p.set_recipient_key(&recipient_key);
                    p.set_channel_id(&[]);
                    p.set_version(1);
                    p.set_seq_up_to(seq_up_to);
                    let mut auth = p.reborrow().init_auth();
                    set_auth(&mut auth)?;
                }
                req.send().promise.await.context("ack RPC failed")?;
                Ok(())
            }
        },
        DEFAULT_MAX_RETRIES,
        DEFAULT_BASE_DELAY_MS,
        anyhow_is_retriable,
    )
    .await
}

/// Fetch multiple peers' hybrid keys in a single round-trip.
/// Returns `None` for peers who have not uploaded a hybrid key.
/// Retries on transient failures with exponential backoff.
pub async fn fetch_hybrid_keys(
    client: &node_service::Client,
    identity_keys: &[&[u8]],
) -> anyhow::Result<Vec<Option<HybridPublicKey>>> {
    let client = client.clone();
    let identity_keys: Vec<Vec<u8>> = identity_keys.iter().map(|k| k.to_vec()).collect();
    retry_async(
        || {
            let client = client.clone();
            let identity_keys = identity_keys.clone();
            async move {
                let mut req = client.fetch_hybrid_keys_request();
                {
                    let mut p = req.get();
                    let mut list = p.reborrow().init_identity_keys(identity_keys.len() as u32);
                    for (i, ik) in identity_keys.iter().enumerate() {
                        list.set(i as u32, ik);
                    }
                    let mut auth = p.reborrow().init_auth();
                    set_auth(&mut auth)?;
                }

                let resp = req
                    .send()
                    .promise
                    .await
                    .context("fetch_hybrid_keys RPC failed")?;

                let keys = resp
                    .get()
                    .context("fetch_hybrid_keys: bad response")?
                    .get_keys()
                    .context("fetch_hybrid_keys: missing keys")?;

                let mut result = Vec::with_capacity(keys.len() as usize);
                for i in 0..keys.len() {
                    let pk_bytes = keys
                        .get(i)
                        .context("fetch_hybrid_keys: key read failed")?
                        .to_vec();
                    if pk_bytes.is_empty() {
                        result.push(None);
                    } else {
                        let pk = HybridPublicKey::from_bytes(&pk_bytes)
                            .context("invalid hybrid public key")?;
                        result.push(Some(pk));
                    }
                }

                Ok(result)
            }
        },
        DEFAULT_MAX_RETRIES,
        DEFAULT_BASE_DELAY_MS,
        anyhow_is_retriable,
    )
    .await
}

/// Enqueue the same payload to multiple recipients in a single round-trip.
/// Returns per-recipient sequence numbers.
/// Retries on transient failures with exponential backoff.
pub async fn batch_enqueue(
    client: &node_service::Client,
    recipient_keys: &[&[u8]],
    payload: &[u8],
) -> anyhow::Result<Vec<u64>> {
    let client = client.clone();
    let recipient_keys: Vec<Vec<u8>> = recipient_keys.iter().map(|k| k.to_vec()).collect();
    let payload = payload.to_vec();
    retry_async(
        || {
            let client = client.clone();
            let recipient_keys = recipient_keys.clone();
            let payload = payload.clone();
            async move {
                let mut req = client.batch_enqueue_request();
                {
                    let mut p = req.get();
                    let mut list = p.reborrow().init_recipient_keys(recipient_keys.len() as u32);
                    for (i, rk) in recipient_keys.iter().enumerate() {
                        list.set(i as u32, rk);
                    }
                    p.set_payload(&payload);
                    p.set_channel_id(&[]);
                    p.set_version(1);
                    let mut auth = p.reborrow().init_auth();
                    set_auth(&mut auth)?;
                }

                let resp = req
                    .send()
                    .promise
                    .await
                    .context("batch_enqueue RPC failed")?;

                let seqs = resp
                    .get()
                    .context("batch_enqueue: bad response")?
                    .get_seqs()
                    .context("batch_enqueue: missing seqs")?;

                let mut result = Vec::with_capacity(seqs.len() as usize);
                for i in 0..seqs.len() {
                    result.push(seqs.get(i));
                }

                Ok(result)
            }
        },
        DEFAULT_MAX_RETRIES,
        DEFAULT_BASE_DELAY_MS,
        anyhow_is_retriable,
    )
    .await
}

/// Resolve a username to its Ed25519 identity key (32 bytes).
///
/// When the server returns a non-empty `inclusionProof`, the client verifies it
/// against the identity key using the Key Transparency Merkle proof. Proof
/// verification failure is treated as a hard error (the server is misbehaving).
/// If the server sends no proof (empty field), the key is returned as-is —
/// callers can decide whether to require proofs for security-critical flows.
///
/// Returns `None` if the username is not registered.
pub async fn resolve_user(
    client: &node_service::Client,
    username: &str,
) -> anyhow::Result<Option<Vec<u8>>> {
    let mut req = client.resolve_user_request();
    {
        let mut p = req.get();
        p.set_username(username);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("resolve_user RPC failed")?;

    let reader = resp.get().context("resolve_user: bad response")?;

    let key = reader
        .get_identity_key()
        .context("resolve_user: missing identity_key field")?
        .to_vec();

    if key.is_empty() {
        return Ok(None);
    }

    // Verify the KT inclusion proof when the server sends one.
    let proof_bytes = reader
        .get_inclusion_proof()
        .context("resolve_user: missing inclusion_proof field")?
        .to_vec();

    if !proof_bytes.is_empty() {
        let proof = quicproquo_kt::InclusionProof::from_bytes(&proof_bytes)
            .context("resolve_user: inclusion proof deserialise failed")?;
        quicproquo_kt::verify_inclusion(&proof, username, &key)
            .context("resolve_user: KT inclusion proof verification FAILED — possible key mislabelling")?;
    }

    Ok(Some(key))
}

/// Reverse lookup: resolve an identity key to the registered username.
/// Returns `None` if no username is associated with the key.
pub async fn resolve_identity(
    client: &node_service::Client,
    identity_key: &[u8],
) -> anyhow::Result<Option<String>> {
    let mut req = client.resolve_identity_request();
    {
        let mut p = req.get();
        p.set_identity_key(identity_key);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("resolve_identity RPC failed")?;

    let username = resp
        .get()
        .context("resolve_identity: bad response")?
        .get_username()
        .context("resolve_identity: missing field")?
        .to_str()
        .unwrap_or("")
        .to_string();

    if username.is_empty() {
        Ok(None)
    } else {
        Ok(Some(username))
    }
}

/// Create a 1:1 DM channel with a peer.
///
/// Returns `(channel_id, was_new)` where `channel_id` is the stable 16-byte identifier and
/// `was_new` is `true` iff this call created the channel for the first time. When `was_new` is
/// `false`, the channel already existed (created by the peer), and the caller should wait for
/// the peer's MLS Welcome to arrive via the background poller rather than creating a new MLS group.
pub async fn create_channel(
    client: &node_service::Client,
    peer_key: &[u8],
) -> anyhow::Result<(Vec<u8>, bool)> {
    let mut req = client.create_channel_request();
    {
        let mut p = req.get();
        p.set_peer_key(peer_key);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("create_channel RPC failed")?;

    let reader = resp.get().context("create_channel: bad response")?;
    let channel_id = reader
        .get_channel_id()
        .context("create_channel: missing channel_id")?
        .to_vec();
    let was_new = reader.get_was_new();

    Ok((channel_id, was_new))
}

/// Upload a single chunk of a blob to the server.
///
/// `blob_hash` is the expected SHA-256 hash (32 bytes) of the complete blob.
/// Returns the `blob_id` once the server has received and verified the final chunk.
pub async fn upload_blob_chunk(
    client: &node_service::Client,
    blob_hash: &[u8],
    chunk: &[u8],
    offset: u64,
    total_size: u64,
    mime_type: &str,
) -> anyhow::Result<Vec<u8>> {
    let mut req = client.upload_blob_request();
    {
        let mut p = req.get();
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
        p.set_blob_hash(blob_hash);
        p.set_chunk(chunk);
        p.set_offset(offset);
        p.set_total_size(total_size);
        p.set_mime_type(mime_type);
    }
    let resp = req.send().promise.await.context("upload_blob RPC failed")?;
    let blob_id = resp
        .get()
        .context("upload_blob: bad response")?
        .get_blob_id()
        .context("upload_blob: missing blob_id")?
        .to_vec();
    Ok(blob_id)
}

/// Download a single chunk of a blob from the server.
///
/// Returns `(chunk_bytes, total_size, mime_type)`.
pub async fn download_blob_chunk(
    client: &node_service::Client,
    blob_id: &[u8],
    offset: u64,
    length: u32,
) -> anyhow::Result<(Vec<u8>, u64, String)> {
    let mut req = client.download_blob_request();
    {
        let mut p = req.get();
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
        p.set_blob_id(blob_id);
        p.set_offset(offset);
        p.set_length(length);
    }
    let resp = req.send().promise.await.context("download_blob RPC failed")?;
    let reader = resp.get().context("download_blob: bad response")?;
    let chunk = reader.get_chunk().context("download_blob: missing chunk")?.to_vec();
    let total_size = reader.get_total_size();
    let mime_type = reader
        .get_mime_type()
        .context("download_blob: missing mime_type")?
        .to_str()
        .unwrap_or("application/octet-stream")
        .to_string();
    Ok((chunk, total_size, mime_type))
}

/// Delete the authenticated user's account on the server.
/// Requires an identity-bound session (OPAQUE login).
pub async fn delete_account(
    client: &node_service::Client,
) -> anyhow::Result<bool> {
    let mut req = client.delete_account_request();
    {
        let mut p = req.get();
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("delete_account RPC failed")?;

    let success = resp
        .get()
        .context("delete_account: bad response")?
        .get_success();

    Ok(success)
}

/// Register a device for the authenticated identity.
pub async fn register_device(
    client: &node_service::Client,
    device_id: &[u8],
    device_name: &str,
) -> anyhow::Result<bool> {
    let mut req = client.register_device_request();
    {
        let mut p = req.get();
        p.set_device_id(device_id);
        p.set_device_name(device_name);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("register_device RPC failed")?;

    let success = resp
        .get()
        .context("register_device: bad response")?
        .get_success();

    Ok(success)
}

/// List all registered devices for the authenticated identity.
pub async fn list_devices(
    client: &node_service::Client,
) -> anyhow::Result<Vec<(Vec<u8>, String, u64)>> {
    let mut req = client.list_devices_request();
    {
        let mut p = req.get();
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("list_devices RPC failed")?;

    let devices = resp
        .get()
        .context("list_devices: bad response")?
        .get_devices()
        .context("list_devices: missing devices field")?;

    let mut result = Vec::with_capacity(devices.len() as usize);
    for i in 0..devices.len() {
        let entry = devices.get(i);
        let device_id = entry
            .get_device_id()
            .context("list_devices: missing device_id")?
            .to_vec();
        let device_name = entry
            .get_device_name()
            .context("list_devices: missing device_name")?
            .to_str()
            .unwrap_or("")
            .to_string();
        let registered_at = entry.get_registered_at();
        result.push((device_id, device_name, registered_at));
    }

    Ok(result)
}

/// Revoke (remove) a registered device.
pub async fn revoke_device(
    client: &node_service::Client,
    device_id: &[u8],
) -> anyhow::Result<bool> {
    let mut req = client.revoke_device_request();
    {
        let mut p = req.get();
        p.set_device_id(device_id);
        let mut auth = p.reborrow().init_auth();
        set_auth(&mut auth)?;
    }

    let resp = req
        .send()
        .promise
        .await
        .context("revoke_device RPC failed")?;

    let success = resp
        .get()
        .context("revoke_device: bad response")?
        .get_success();

    Ok(success)
}

/// Return the current Unix timestamp in milliseconds.
pub fn current_timestamp_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}