//! Sharing: per-node share membership, public links, the invitations this
//! account has received, saved public-link bookmarks, and the reverse view of
//! everything this account shares out.
//!
//! All of it is remote-only state — nothing here touches the inode maps or the
//! content cache, so every method is a round-trip wrapped in the crate error
//! shape the control socket expects.

use std::path::Path;

use pdfs_core::control::{
    BookmarkInfo, DirEntry, InvitationInfo, LinkChange, PublicLinkInfo, ShareEntry, ShareEntryKind,
    SharedItem,
};
use pdfs_core::{CoreError, CoreResult};
use tracing::debug;

use crate::QUOTA_TTL;
use proton_drive_rs::proton_sdk::crypto::VerificationStatus;
use proton_drive_rs::proton_sdk::ids::NodeUid;
use proton_drive_rs::{MemberRole, NodeKind};

use proton_drive_rs::Node;

use super::{Core, node_size, parse_uid, public_link_info, role_from_str, role_to_str};

/// A node from a *shared* listing as a [`DirEntry`].
///
/// Still uid-addressed — that is the only handle a foreign-volume node is
/// guaranteed to have — but `path` is filled in when the node is resident under
/// the synthetic `Shared with me/` directory, which is what lets a front-end
/// offer path-keyed actions on it. The local-state flags (`pinned`, `cached`)
/// stay false: they describe content this account owns a local copy policy for,
/// which a share on someone else's volume is not.
fn shared_entry_of(core: &Core, n: Node) -> DirEntry {
    let is_dir = n.is_folder();
    let size = match &n.kind {
        NodeKind::File {
            claimed_size,
            total_size_on_storage,
            ..
        } => claimed_size.unwrap_or(*total_size_on_storage).max(0) as u64,
        NodeKind::Folder => 0,
    };
    let role = n
        .membership
        .as_ref()
        .and_then(|m| m.role_exact())
        .map(role_to_str)
        .unwrap_or_default()
        .to_string();
    let invitation = n.membership.as_ref();
    let shared_by = invitation
        .and_then(|m| m.inviter_email.clone())
        .unwrap_or_default();
    let shared_at = invitation.and_then(|m| m.invite_time).unwrap_or(0);
    // Only a signature that is present and wrong is a warning. An invitation
    // that could not be checked — unsigned, or no keys for the claimed inviter —
    // is not evidence of forgery, and flagging every such share would teach the
    // user to ignore the flag.
    let shared_by_unverified = invitation
        .and_then(|m| m.inviter_verification)
        .is_some_and(|status| status == VerificationStatus::Failed);
    DirEntry {
        name: n.name,
        is_dir,
        size,
        modified: n.modification_time,
        pinned: false,
        pinned_by_folder: false,
        cached: false,
        path: core.rel_path_for_uid(&n.uid).unwrap_or_default(),
        uid: n.uid.to_string(),
        role,
        shared_by,
        shared_at,
        shared_by_unverified,
        trashed_at: 0,
        trashed_from: None,
        issue: None,
    }
}

impl Core {
    // ---- sharing a node ---------------------------------------------------

    /// Invite Proton and/or external emails to the node at `rel` at `role`.
    /// Returns `(proton_invited, external_invited)`.
    pub(crate) fn share_node(
        &self,
        rel: &Path,
        emails: &[String],
        role: &str,
        message: Option<&str>,
    ) -> CoreResult<(usize, usize)> {
        let (_ino, uid) = self.resolve(rel)?;
        self.share_node_for_uid(&uid, emails, role, message)
    }

    /// [`Core::share_node`] for a node addressed through any daemon location.
    pub(crate) fn share_node_by_uid(
        &self,
        uid: &str,
        emails: &[String],
        role: &str,
        message: Option<&str>,
    ) -> CoreResult<(usize, usize)> {
        let uid = self.resolve_anywhere(uid)?;
        self.share_node_for_uid(&uid, emails, role, message)
    }

    fn share_node_for_uid(
        &self,
        uid: &NodeUid,
        emails: &[String],
        role: &str,
        message: Option<&str>,
    ) -> CoreResult<(usize, usize)> {
        let role = role_from_str(role)?;
        let invitees: Vec<(String, MemberRole)> =
            emails.iter().map(|e| (e.clone(), role)).collect();
        self.rt
            .block_on(self.client.invite_users(uid, &invitees, message))
            .map_err(|e| CoreError::from_api(&e, "share"))
    }

    /// List the members, pending invitations and public link of the node at `rel`.
    pub(crate) fn list_share(
        &self,
        rel: &Path,
    ) -> CoreResult<(Vec<ShareEntry>, Option<PublicLinkInfo>)> {
        let (_ino, uid) = self.resolve(rel)?;
        self.list_share_for_uid(&uid)
    }

    /// [`Core::list_share`] for a node addressed through any daemon location.
    pub(crate) fn list_share_by_uid(
        &self,
        uid: &str,
    ) -> CoreResult<(Vec<ShareEntry>, Option<PublicLinkInfo>)> {
        let uid = self.resolve_anywhere(uid)?;
        self.list_share_for_uid(&uid)
    }

    fn list_share_for_uid(
        &self,
        uid: &NodeUid,
    ) -> CoreResult<(Vec<ShareEntry>, Option<PublicLinkInfo>)> {
        let mut entries = Vec::new();
        for m in self
            .rt
            .block_on(self.client.list_share_members(uid))
            .map_err(|e| CoreError::from_api(&e, "list members"))?
        {
            entries.push(ShareEntry {
                id: m.membership_id.to_string(),
                email: m.email,
                role: role_to_str(m.role).to_string(),
                kind: ShareEntryKind::Member,
            });
        }
        for inv in self
            .rt
            .block_on(self.client.list_share_invitations(uid))
            .map_err(|e| CoreError::from_api(&e, "list invitations"))?
        {
            entries.push(ShareEntry {
                id: inv.invitation_id,
                email: inv.invitee_email,
                role: role_to_str(inv.role).to_string(),
                kind: ShareEntryKind::ProtonInvite,
            });
        }
        for ext in self
            .rt
            .block_on(self.client.list_external_invitations(uid))
            .map_err(|e| CoreError::from_api(&e, "list external invitations"))?
        {
            entries.push(ShareEntry {
                id: ext.invitation_id,
                email: ext.invitee_email,
                role: role_to_str(ext.role).to_string(),
                kind: ShareEntryKind::ExternalInvite,
            });
        }

        let link = self
            .rt
            .block_on(self.client.get_public_link(uid))
            .map_err(|e| CoreError::from_api(&e, "get public link"))?
            .map(public_link_info);

        Ok((entries, link))
    }

    /// Change the role of a member or pending Proton invitation on the node at
    /// `rel`. External invitations have no role-update endpoint.
    pub(crate) fn update_share_role(
        &self,
        rel: &Path,
        id: &str,
        kind: ShareEntryKind,
        role: &str,
    ) -> CoreResult<()> {
        let (_ino, uid) = self.resolve(rel)?;
        self.update_share_role_for_uid(&uid, id, kind, role)
    }

    /// [`Core::update_share_role`] for a node addressed through any daemon
    /// location.
    pub(crate) fn update_share_role_by_uid(
        &self,
        uid: &str,
        id: &str,
        kind: ShareEntryKind,
        role: &str,
    ) -> CoreResult<()> {
        let uid = self.resolve_anywhere(uid)?;
        self.update_share_role_for_uid(&uid, id, kind, role)
    }

    fn update_share_role_for_uid(
        &self,
        uid: &NodeUid,
        id: &str,
        kind: ShareEntryKind,
        role: &str,
    ) -> CoreResult<()> {
        let role = role_from_str(role)?;
        match kind {
            ShareEntryKind::Member => {
                let member = self
                    .rt
                    .block_on(self.client.list_share_members(uid))
                    .map_err(|e| CoreError::from_api(&e, "list members"))?
                    .into_iter()
                    .find(|m| m.membership_id.to_string() == id)
                    .ok_or_else(|| CoreError::not_found("member not found"))?;
                self.rt
                    .block_on(self.client.update_member_role(&member, role))
                    .map_err(|e| CoreError::from_api(&e, "update role"))
            }
            ShareEntryKind::ProtonInvite => {
                let inv = self
                    .rt
                    .block_on(self.client.list_share_invitations(uid))
                    .map_err(|e| CoreError::from_api(&e, "list invitations"))?
                    .into_iter()
                    .find(|i| i.invitation_id == id)
                    .ok_or_else(|| CoreError::not_found("invitation not found"))?;
                self.rt
                    .block_on(self.client.update_invitation_role(&inv, role))
                    .map_err(|e| CoreError::from_api(&e, "update role"))
            }
            ShareEntryKind::ExternalInvite => Err(CoreError::invalid(
                "an external invitation's role cannot be changed",
            )),
        }
    }

    /// Remove a member, pending Proton invite, or external invite from the node
    /// at `rel`.
    pub(crate) fn remove_share_entry(
        &self,
        rel: &Path,
        id: &str,
        kind: ShareEntryKind,
    ) -> CoreResult<()> {
        let (_ino, uid) = self.resolve(rel)?;
        self.remove_share_entry_for_uid(&uid, id, kind)
    }

    /// [`Core::remove_share_entry`] for a node addressed through any daemon
    /// location.
    pub(crate) fn remove_share_entry_by_uid(
        &self,
        uid: &str,
        id: &str,
        kind: ShareEntryKind,
    ) -> CoreResult<()> {
        let uid = self.resolve_anywhere(uid)?;
        self.remove_share_entry_for_uid(&uid, id, kind)
    }

    fn remove_share_entry_for_uid(
        &self,
        uid: &NodeUid,
        id: &str,
        kind: ShareEntryKind,
    ) -> CoreResult<()> {
        match kind {
            ShareEntryKind::Member => {
                let member = self
                    .rt
                    .block_on(self.client.list_share_members(uid))
                    .map_err(|e| CoreError::from_api(&e, "list members"))?
                    .into_iter()
                    .find(|m| m.membership_id.to_string() == id)
                    .ok_or_else(|| CoreError::not_found("member not found"))?;
                self.rt
                    .block_on(self.client.remove_member(&member))
                    .map_err(|e| CoreError::from_api(&e, "remove member"))
            }
            ShareEntryKind::ProtonInvite => {
                let inv = self
                    .rt
                    .block_on(self.client.list_share_invitations(uid))
                    .map_err(|e| CoreError::from_api(&e, "list invitations"))?
                    .into_iter()
                    .find(|i| i.invitation_id == id)
                    .ok_or_else(|| CoreError::not_found("invitation not found"))?;
                self.rt
                    .block_on(self.client.delete_invitation(&inv))
                    .map_err(|e| CoreError::from_api(&e, "revoke invitation"))
            }
            ShareEntryKind::ExternalInvite => {
                let ext = self
                    .rt
                    .block_on(self.client.list_external_invitations(uid))
                    .map_err(|e| CoreError::from_api(&e, "list external invitations"))?
                    .into_iter()
                    .find(|i| i.invitation_id == id)
                    .ok_or_else(|| CoreError::not_found("external invitation not found"))?;
                self.rt
                    .block_on(self.client.delete_external_invitation(&ext))
                    .map_err(|e| CoreError::from_api(&e, "revoke external invitation"))
            }
        }
    }

    /// Create a public link on the node at `rel`, returning its info (with URL).
    pub(crate) fn create_public_link(
        &self,
        rel: &Path,
        role: &str,
        password: Option<&str>,
        expires: Option<i64>,
    ) -> CoreResult<PublicLinkInfo> {
        let (_ino, uid) = self.resolve(rel)?;
        self.create_public_link_for_uid(&uid, role, password, expires)
    }

    /// [`Core::create_public_link`] for a node addressed through any daemon
    /// location.
    pub(crate) fn create_public_link_by_uid(
        &self,
        uid: &str,
        role: &str,
        password: Option<&str>,
        expires: Option<i64>,
    ) -> CoreResult<PublicLinkInfo> {
        let uid = self.resolve_anywhere(uid)?;
        self.create_public_link_for_uid(&uid, role, password, expires)
    }

    fn create_public_link_for_uid(
        &self,
        uid: &NodeUid,
        role: &str,
        password: Option<&str>,
        expires: Option<i64>,
    ) -> CoreResult<PublicLinkInfo> {
        let role = role_from_str(role)?;
        let link = self
            .rt
            .block_on(self.client.create_public_link(uid, role, password, expires))
            .map_err(|e| CoreError::from_api(&e, "create public link"))?;
        Ok(public_link_info(link))
    }

    /// Change the public link `id` on the node at `rel`, keeping its URL.
    pub(crate) fn update_public_link(
        &self,
        rel: &Path,
        id: &str,
        role: Option<&str>,
        password: LinkChange<String>,
        expires: LinkChange<i64>,
    ) -> CoreResult<PublicLinkInfo> {
        let (_ino, uid) = self.resolve(rel)?;
        self.update_public_link_for_uid(&uid, id, role, password, expires)
    }

    /// [`Core::update_public_link`] for a node addressed through any daemon
    /// location.
    pub(crate) fn update_public_link_by_uid(
        &self,
        uid: &str,
        id: &str,
        role: Option<&str>,
        password: LinkChange<String>,
        expires: LinkChange<i64>,
    ) -> CoreResult<PublicLinkInfo> {
        let uid = self.resolve_anywhere(uid)?;
        self.update_public_link_for_uid(&uid, id, role, password, expires)
    }

    /// The SDK sets all three options on every update, so each `Keep` is
    /// filled in from the link as it stands.
    fn update_public_link_for_uid(
        &self,
        uid: &NodeUid,
        id: &str,
        role: Option<&str>,
        password: LinkChange<String>,
        expires: LinkChange<i64>,
    ) -> CoreResult<PublicLinkInfo> {
        let link = self
            .rt
            .block_on(self.client.get_public_link(uid))
            .map_err(|e| CoreError::from_api(&e, "get public link"))?
            .filter(|l| l.public_link_id == id)
            .ok_or_else(|| CoreError::not_found("public link not found"))?;
        let role = match role {
            Some(role) => role_from_str(role)?,
            None => link.role,
        };
        let unchanged = role == link.role
            && matches!(password, LinkChange::Keep)
            && matches!(expires, LinkChange::Keep);
        if unchanged {
            return Ok(public_link_info(link));
        }
        let password = match password {
            LinkChange::Keep if link.has_custom_password => {
                // Sending `None` here would quietly remove the password.
                Some(link.custom_password.clone().ok_or_else(|| {
                    CoreError::conflict(
                        "the link's password could not be read; set a new one or remove it",
                    )
                })?)
            }
            LinkChange::Keep | LinkChange::Remove => None,
            LinkChange::Set(password) if password.is_empty() => {
                return Err(CoreError::invalid("the new password is empty"));
            }
            LinkChange::Set(password) => Some(password),
        };
        let expires = match expires {
            LinkChange::Keep => link.expiration_time,
            LinkChange::Remove => None,
            LinkChange::Set(at) => Some(at),
        };
        let link = self
            .rt
            .block_on(
                self.client
                    .update_public_link(uid, role, password.as_deref(), expires),
            )
            .map_err(|e| CoreError::from_api(&e, "update public link"))?;
        Ok(public_link_info(link))
    }

    /// Remove the public link `id` from the node at `rel`.
    pub(crate) fn remove_public_link(&self, rel: &Path, id: &str) -> CoreResult<()> {
        let (_ino, uid) = self.resolve(rel)?;
        self.remove_public_link_for_uid(&uid, id)
    }

    /// [`Core::remove_public_link`] for a node addressed through any daemon
    /// location.
    pub(crate) fn remove_public_link_by_uid(&self, uid: &str, id: &str) -> CoreResult<()> {
        let uid = self.resolve_anywhere(uid)?;
        self.remove_public_link_for_uid(&uid, id)
    }

    fn remove_public_link_for_uid(&self, uid: &NodeUid, id: &str) -> CoreResult<()> {
        let link = self
            .rt
            .block_on(self.client.get_public_link(uid))
            .map_err(|e| CoreError::from_api(&e, "get public link"))?
            .filter(|l| l.public_link_id == id)
            .ok_or_else(|| CoreError::not_found("public link not found"))?;
        self.rt
            .block_on(self.client.remove_public_link(&link))
            .map_err(|e| CoreError::from_api(&e, "remove public link"))
    }

    /// Stop sharing the node at `rel`: remove its public link, then every
    /// external invitation, Proton invitation and member. The SDK has no call
    /// that deletes a share outright, so this composes the per-entry ones.
    pub(crate) fn stop_sharing(&self, rel: &Path) -> CoreResult<StopSharingOutcome> {
        let (_ino, uid) = self.resolve(rel)?;
        self.stop_sharing_for_uid(&uid)
    }

    /// [`Core::stop_sharing`] for a node addressed through any daemon location.
    pub(crate) fn stop_sharing_by_uid(&self, uid: &str) -> CoreResult<StopSharingOutcome> {
        let uid = self.resolve_anywhere(uid)?;
        self.stop_sharing_for_uid(&uid)
    }

    /// Each step runs even when an earlier one failed, so a single stuck
    /// member does not leave the public link up. The link goes first: it is the
    /// widest exposure. Only a failure to list what is there fails the whole
    /// request; a failed removal is counted in the outcome.
    fn stop_sharing_for_uid(&self, uid: &NodeUid) -> CoreResult<StopSharingOutcome> {
        let mut outcome = StopSharingOutcome::default();
        let link = self
            .rt
            .block_on(self.client.get_public_link(uid))
            .map_err(|e| CoreError::from_api(&e, "get public link"))?;
        if let Some(link) = link {
            let result = self.rt.block_on(self.client.remove_public_link(&link));
            outcome.record("the public link", result);
        }
        let external = self
            .rt
            .block_on(self.client.list_external_invitations(uid))
            .map_err(|e| CoreError::from_api(&e, "list external invitations"))?;
        for ext in external {
            let result = self
                .rt
                .block_on(self.client.delete_external_invitation(&ext));
            outcome.record(&ext.invitee_email, result);
        }
        let invitations = self
            .rt
            .block_on(self.client.list_share_invitations(uid))
            .map_err(|e| CoreError::from_api(&e, "list invitations"))?;
        for inv in invitations {
            let result = self.rt.block_on(self.client.delete_invitation(&inv));
            outcome.record(&inv.invitee_email, result);
        }
        let members = self
            .rt
            .block_on(self.client.list_share_members(uid))
            .map_err(|e| CoreError::from_api(&e, "list members"))?;
        for member in members {
            let result = self.rt.block_on(self.client.remove_member(&member));
            outcome.record(&member.email, result);
        }
        if !outcome.failed.is_empty() {
            debug!(%uid, failed = ?outcome.failed, "stop sharing: some entries remain");
        }
        Ok(outcome)
    }

    // ---- shared with me ---------------------------------------------------

    /// List nodes shared with me that I have accepted.
    pub(crate) fn list_shared_with_me(&self) -> CoreResult<Vec<DirEntry>> {
        let uids = self
            .rt
            .block_on(self.client.enumerate_shared_with_me_node_uids())
            .map_err(|e| CoreError::from_api(&e, "enumerate shared"))?;
        if uids.is_empty() {
            return Ok(Vec::new());
        }
        let nodes = self
            .rt
            .block_on(self.drive.enumerate_nodes(&uids))
            .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?;
        Ok(nodes
            .into_iter()
            .map(|node| shared_entry_of(self, node))
            .collect())
    }

    /// List the children of a folder shared with me, addressed by uid.
    ///
    /// A shared subtree has no path in this account's mount — the FUSE tree only
    /// spans my own volume — so browsing into one is uid-addressed the whole way
    /// down, each listing handing the front-end the uids it descends with next.
    pub(crate) fn list_shared_folder(&self, uid: &str) -> CoreResult<Vec<DirEntry>> {
        let uid =
            parse_uid(uid).ok_or_else(|| CoreError::invalid(format!("invalid uid: {uid}")))?;
        let child_uids = self
            .rt
            .block_on(self.drive.enumerate_folder_children_node_uids(&uid))
            .map_err(|e| CoreError::from_api(&e, "enumerate shared children"))?;
        if child_uids.is_empty() {
            return Ok(Vec::new());
        }
        let nodes = self
            .rt
            .block_on(self.drive.enumerate_nodes(&child_uids))
            .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?;
        Ok(nodes
            .into_iter()
            .map(|node| shared_entry_of(self, node))
            .collect())
    }

    /// Download a file shared with me into the content cache, returning its
    /// on-disk path (served from cache when a fresh blob already exists).
    ///
    /// The by-uid twin of [`Core::open_file`]: same cache and same tracked
    /// download, but the node is fetched from the server rather than resolved
    /// through the inode tree, which does not cover other people's volumes.
    ///
    /// [`Core::open_file`]: super::Core::open_file
    pub(crate) fn open_shared_file(&self, uid: &str) -> CoreResult<std::path::PathBuf> {
        let uid =
            parse_uid(uid).ok_or_else(|| CoreError::invalid(format!("invalid uid: {uid}")))?;
        let node = self
            .rt
            .block_on(self.drive.get_node(&uid))
            .map_err(|e| CoreError::from_api(&e, "get shared node"))?
            .ok_or_else(|| CoreError::not_found("shared node not found"))?;
        if !node.is_file() {
            return Err(CoreError::invalid("not a regular file"));
        }
        let size = node_size(&node);
        let mtime = node.modification_time;
        if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) {
            return Ok(p);
        }
        let bytes = self
            .download_file_tracked(&uid, &node.name, size)
            .map_err(|e| CoreError::from_api(&e, "download"))?;
        self.cache
            .store(&uid, mtime, size, &bytes)
            .map_err(|e| CoreError::internal(format!("cache store: {e}")))?;
        Ok(self.cache.content_path(&uid))
    }

    /// Leave a shared node by its uid.
    pub(crate) fn leave_shared(&self, uid: &str) -> CoreResult<()> {
        let uid =
            parse_uid(uid).ok_or_else(|| CoreError::invalid(format!("invalid uid: {uid}")))?;
        self.rt
            .block_on(self.client.leave_shared_node(&uid))
            .map_err(|e| CoreError::from_api(&e, "leave shared"))
    }

    // ---- incoming invitations ---------------------------------------------

    /// List invitations addressed to me, pending accept or reject.
    pub(crate) fn list_invitations(&self) -> CoreResult<Vec<InvitationInfo>> {
        let invitations = self
            .rt
            .block_on(self.client.list_incoming_invitations())
            .map_err(|e| CoreError::from_api(&e, "list invitations"))?;
        Ok(invitations
            .into_iter()
            .map(|i| InvitationInfo {
                id: i.invitation_id,
                inviter_email: i.inviter_email,
                name: i.node_name,
                is_dir: i.is_folder,
            })
            .collect())
    }

    /// Accept an invitation addressed to me by its id.
    pub(crate) fn accept_invitation(&self, id: &str) -> CoreResult<()> {
        self.rt
            .block_on(self.client.accept_invitation(id))
            .map_err(|e| CoreError::from_api(&e, "accept invitation"))
    }

    /// Reject an invitation addressed to me by its id.
    pub(crate) fn reject_invitation(&self, id: &str) -> CoreResult<()> {
        self.rt
            .block_on(self.client.reject_invitation(id))
            .map_err(|e| CoreError::from_api(&e, "reject invitation"))
    }

    // ---- bookmarks --------------------------------------------------------

    /// List public links saved to my account.
    pub(crate) fn list_bookmarks(&self) -> CoreResult<Vec<BookmarkInfo>> {
        let bookmarks = self
            .rt
            .block_on(self.client.list_bookmarks())
            .map_err(|e| CoreError::from_api(&e, "list bookmarks"))?;
        Ok(bookmarks
            .into_iter()
            .map(|b| BookmarkInfo {
                token: b.token,
                url: b.url,
                name: b.node_name,
                is_dir: b.is_folder,
            })
            .collect())
    }

    /// Save a public link (optionally password-protected) as a bookmark.
    pub(crate) fn create_bookmark(&self, url: &str, password: Option<&str>) -> CoreResult<()> {
        self.rt
            .block_on(self.client.create_bookmark(url, password))
            .map_err(|e| CoreError::from_api(&e, "create bookmark"))
    }

    /// Remove a saved bookmark by its token.
    pub(crate) fn delete_bookmark(&self, token: &str) -> CoreResult<()> {
        self.rt
            .block_on(self.client.delete_bookmark(token))
            .map_err(|e| CoreError::from_api(&e, "delete bookmark"))
    }

    // ---- account ----------------------------------------------------------

    /// Total account storage usage `(max_space, used_space)` in bytes, across all
    /// Proton products (not Drive-only). A remote round-trip; nothing here is
    /// cached.
    pub(crate) fn account_quota(&self) -> CoreResult<(i64, i64)> {
        let q = self.block_on_bounded(self.drive.quota()).map_err(|e| {
            self.lost_link(&e, "account quota");
            CoreError::from_api(&e, "account quota")
        })?;
        *self.quota.lock() = Some((std::time::Instant::now(), q.max_space, q.used_space));
        Ok((q.max_space, q.used_space))
    }

    /// The account quota as `statfs(2)` needs it: cached for [`QUOTA_TTL`],
    /// refetched after that, and `None` when there is no usable answer at all —
    /// offline with nothing cached.
    ///
    /// Never blocks on the network for a *fresh* value when a recent one exists,
    /// because the callers are `df` and every "is there room?" preflight, and a
    /// stale byte count is worth more to them than a stalled syscall. Offline,
    /// the last known figures keep being served rather than degrading to zero.
    pub(crate) fn account_quota_cached(&self) -> Option<(i64, i64)> {
        if let Some((at, max_space, used_space)) = *self.quota.lock()
            && at.elapsed() < QUOTA_TTL
        {
            return Some((max_space, used_space));
        }
        let last = || self.quota.lock().map(|(_, max, used)| (max, used));
        if !self.is_online()
            || self
                .quota_refreshing
                .swap(true, std::sync::atomic::Ordering::AcqRel)
        {
            return last();
        }
        let refreshed = self.account_quota();
        self.quota_refreshing
            .store(false, std::sync::atomic::Ordering::Release);
        match refreshed {
            Ok(q) => Some(q),
            Err(error) => {
                debug!(%error, "account quota refresh failed; serving the last known figures");
                last()
            }
        }
    }

    // ---- shared by me -----------------------------------------------------

    /// List the nodes I have shared with others, each with a summary of its share
    /// state (members, pending invitations, public link). One list call enumerates
    /// the shared uids; the per-node detail is then gathered best-effort — a single
    /// node racing with an unshare drops from the list rather than failing the whole
    /// request.
    ///
    /// A share with no member, no invitation and no link is left out: it is
    /// shared with no one. Stop Sharing leaves exactly that behind, since the SDK
    /// cannot delete a share, and Drive keeps listing it until its backend gets
    /// round to removing it, minutes later. A share whose details could not all
    /// be read stays listed, so a failed lookup never hides a live share.
    pub(crate) fn list_shared_by_me(&self) -> CoreResult<Vec<SharedItem>> {
        let uids = self
            .rt
            .block_on(self.client.enumerate_shared_by_me_node_uids())
            .map_err(|e| CoreError::from_api(&e, "enumerate shared-by-me"))?;
        if uids.is_empty() {
            return Ok(Vec::new());
        }
        let nodes = self
            .rt
            .block_on(self.drive.enumerate_nodes(&uids))
            .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?;
        let mut items = Vec::with_capacity(nodes.len());
        for n in nodes {
            let uid = n.uid.clone();
            let members = self
                .rt
                .block_on(self.client.list_share_members(&uid))
                .map(|m| m.len());
            let proton_invites = self
                .rt
                .block_on(self.client.list_share_invitations(&uid))
                .map(|i| i.len());
            let external_invites = self
                .rt
                .block_on(self.client.list_external_invitations(&uid))
                .map(|i| i.len());
            let link = self.rt.block_on(self.client.get_public_link(&uid));
            if let (Ok(0), Ok(0), Ok(0), Ok(None)) =
                (&members, &proton_invites, &external_invites, &link)
            {
                debug!(%uid, "shared by me: skipping an emptied share");
                continue;
            }
            let members = members.unwrap_or(0);
            let proton_invites = proton_invites.unwrap_or(0);
            let external_invites = external_invites.unwrap_or(0);
            let link = link.ok().flatten().map(public_link_info);
            items.push(SharedItem {
                uid: uid.to_string(),
                is_dir: n.is_folder(),
                modified: n.modification_time,
                name: n.name,
                path: self.rel_path_for_uid(&uid).unwrap_or_default(),
                member_count: members,
                invite_count: proton_invites + external_invites,
                link,
            });
        }
        Ok(items)
    }

    /// Best-effort mountpoint-relative path for a node already interned in the live
    /// tree, by walking parent inodes to the root. `None` when the node has never
    /// been seen through the mount (e.g. shared but not browsed to this session) —
    /// the caller then leaves the path empty.
    pub(crate) fn rel_path_for_uid(&self, uid: &NodeUid) -> Option<String> {
        // Relative to whichever root the node sits under: a node browsed
        // through a sync folder is under that fork's root, not My Files
        // (`docs/BUGS.md` B74).
        walk_to_root(&self.state(), uid)
    }
}

/// What [`Core::stop_sharing`] managed to remove.
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct StopSharingOutcome {
    /// Entries removed (the link, invitations and members alike).
    pub(crate) removed: usize,
    /// What could not be removed, by email (or "the public link"), with why.
    pub(crate) failed: Vec<(String, String)>,
}

impl StopSharingOutcome {
    fn record<E: std::fmt::Display>(&mut self, what: &str, result: Result<(), E>) {
        match result {
            Ok(()) => self.removed += 1,
            Err(e) => self.failed.push((what.to_string(), e.to_string())),
        }
    }

    /// The reply for the control socket: a sentence for a full success, an
    /// error naming what is left otherwise.
    pub(crate) fn into_result(self) -> Result<String, String> {
        if self.failed.is_empty() {
            return Ok(match self.removed {
                0 => "it was not shared".to_string(),
                1 => "stopped sharing: removed 1 entry".to_string(),
                n => format!("stopped sharing: removed {n} entries"),
            });
        }
        let left = self
            .failed
            .iter()
            .map(|(what, why)| format!("{what} ({why})"))
            .collect::<Vec<_>>()
            .join(", ");
        Err(format!(
            "removed {} of {}; still shared: {left}",
            self.removed,
            self.removed + self.failed.len()
        ))
    }
}

/// Walk the tree from `uid` up to the root it sits under, building the path.
/// `None` if the node has never been interned, or if the chain breaks partway —
/// a broken chain is a missing path, never a partial one.
fn walk_to_root(st: &super::State, uid: &NodeUid) -> Option<String> {
    let mut ino = *st.by_uid.get(uid)?;
    let mut parts = Vec::new();
    while !st.roots.contains(&ino) {
        let entry = st.entries.get(&ino)?;
        // An orphan is its own parent without being a root.
        if entry.parent == ino {
            return None;
        }
        parts.push(entry.node.name.clone());
        ino = entry.parent;
    }
    parts.reverse();
    Some(parts.join("/"))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn stopping_a_share_that_is_gone_says_so() {
        let outcome = StopSharingOutcome::default();
        assert_eq!(outcome.into_result().unwrap(), "it was not shared");
    }

    #[test]
    fn a_failed_removal_names_what_is_still_shared() {
        let mut outcome = StopSharingOutcome::default();
        outcome.record::<String>("the public link", Ok(()));
        outcome.record("bob@example.com", Err("forbidden"));
        outcome.record::<String>("carol@example.com", Ok(()));
        assert_eq!(
            outcome.into_result().unwrap_err(),
            "removed 2 of 3; still shared: bob@example.com (forbidden)"
        );
    }

    #[test]
    fn a_full_removal_counts_the_entries() {
        let mut outcome = StopSharingOutcome::default();
        outcome.record::<String>("the public link", Ok(()));
        outcome.record::<String>("bob@example.com", Ok(()));
        assert_eq!(
            outcome.into_result().unwrap(),
            "stopped sharing: removed 2 entries"
        );
    }
}