gunnar_server/control_plane/

clan_plane.rs

//! **Clans on the real plane** — [`gunnar_grpc::clans::ClanPlane`] for
//! [`CoreControlPlane`] (owner, 2026-10-05). The clan rows are
//! `gunnar_core::AuthTable`'s, in one compare-and-swap with the grants rule 2½
//! widens; the clan's namespace is the catalog's, claimed FIRST — one claim for
//! the name, exactly as a principal's account namespace is claimed with it —
//! and given back if the table refuses the clan.
//!
//! **Round 2** (owner, 2026-10-05): a member whose paid period has ended
//! LAPSES — its share stays on the row and the door stops counting it
//! ([`CoreControlPlane::clan_lapsed`], the entitlement slot's one reading);
//! and a repository moves OUT of a clan into an account, offered by the
//! clan's admin and taken by the account's admin, through the write paths
//! every push takes ([`CoreControlPlane::move_repository`]) so the twin and
//! a restore see it as they see everything else.

use std::sync::Arc;

use anyhow::Context as _;
use gunnar_core::{AccountRoom, ClanRefusal, ClanRole};
use gunnar_grpc::clans::{ClanPlane, ClanRoom};
use gunnar_grpc::seam::{ControlError, ControlResult, DeleteOptions};
use gunnar_grpc::ControlPlane as _;
use gunnar_lfs::LfsStore as _;
use gunnar_store::git_storage_trait::{Caps, RefUpdate};

use super::{internal, CoreControlPlane};

/// The table's clan onto the wire: what the plane knows. The service fills
/// what the caller sees of it. `lapsed`: the members whose shares are out of
/// force now.
fn clan_out(clan: &gunnar_core::Clan, lapsed: &[String]) -> gunnar_proto::Clan {
    gunnar_proto::Clan {
        name: clan.name.clone(),
        created_by: clan.created_by.clone(),
        created_at_unix_ms: clan.created_at,
        members: clan
            .members
            .iter()
            .map(|m| gunnar_proto::ClanMember {
                principal_id: m.principal.clone(),
                admin: m.admin,
                contribution_bytes: m.contribution_bytes,
                joined_at_unix_ms: m.joined_at,
                lapsed: lapsed.contains(&m.principal),
            })
            .collect(),
        invites: clan
            .invites
            .iter()
            .map(|i| gunnar_proto::ClanInvite {
                principal_id: i.principal.clone(),
                invited_by: i.by.clone(),
                at_unix_ms: i.at,
            })
            .collect(),
        room_bytes: clan.room_bytes(),
        room_in_force_bytes: clan.room_in_force(lapsed),
        offers: clan.offers.iter().map(|o| offer_out(&clan.name, o, None)).collect(),
        ..Default::default()
    }
}

/// An offer onto the wire.
fn offer_out(clan: &str, offer: &gunnar_core::ClanRepoOffer, stored: Option<u64>) -> gunnar_proto::RepoOffer {
    gunnar_proto::RepoOffer {
        clan: clan.to_owned(),
        repo: offer.repo.clone(),
        to_account: offer.to.clone(),
        offered_by: offer.by.clone(),
        at_unix_ms: offer.at,
        stored_bytes: stored,
    }
}

fn room_in(room: ClanRoom) -> AccountRoom {
    AccountRoom {
        cap: room.cap,
        stored: room.stored,
    }
}

/// A refusal the table made, as the seam says it.
fn clan_error(err: gunnar_core::Error) -> ControlError {
    match err {
        gunnar_core::Error::Clan {
            clan,
            refusal,
            reason,
        } => match refusal {
            ClanRefusal::NotFound => {
                let _ = reason;
                ControlError::NotFound(format!("clan {clan} (none you are in or invited to)"))
            }
            ClanRefusal::Denied => ControlError::PermissionDenied(reason),
            ClanRefusal::Precondition => ControlError::FailedPrecondition(reason),
            ClanRefusal::Invalid => ControlError::InvalidArgument(reason),
            ClanRefusal::Exists => ControlError::AlreadyExists(reason),
        },
        gunnar_core::Error::PrincipalExists(id) => ControlError::AlreadyExists(format!("an account called {id}")),
        other => internal(other),
    }
}

impl CoreControlPlane {
    /// After a clan write: the identity half of the metadata mirror, as
    /// every other identity write does.
    fn clan_written(&self, clan: gunnar_core::Clan) -> gunnar_proto::Clan {
        self.stores.mirror_identity();
        self.clan_view(&clan)
    }

    /// The clan onto the wire, its lapsed shares and its offers' sizes read.
    fn clan_view(&self, clan: &gunnar_core::Clan) -> gunnar_proto::Clan {
        let mut out = clan_out(clan, &self.clan_lapsed(clan));
        if !out.offers.is_empty() {
            for offer in &mut out.offers {
                offer.stored_bytes = self.repo_stored(&clan.name, &offer.repo);
            }
        }
        out
    }

    /// **The members whose shares are out of force now** (round 2, owner
    /// 2026-10-05: "the clan KEEPS that member's room share until the
    /// member's paid period ends, then the share drops"). The entitlement
    /// slot reads each member's fact — the one reading the door makes too
    /// ([`crate::entitlement::EntitlementSlot::lapsed_members`]). A box that
    /// sells nothing holds no paid period, and no share lapses there.
    pub(crate) fn clan_lapsed(&self, clan: &gunnar_core::Clan) -> Vec<String> {
        #[cfg(feature = "metered")]
        if let Some(slot) = self.entitlement.as_ref() {
            return slot.lapsed_members(clan);
        }
        let _ = clan;
        Vec::new()
    }

    /// [`Self::clan_lapsed`] by the clan's name (empty for no clan).
    fn lapsed_in(&self, clan: &str) -> ControlResult<Vec<String>> {
        Ok(self
            .stores
            .auth()
            .clan(clan)
            .map_err(internal)?
            .map(|row| self.clan_lapsed(&row))
            .unwrap_or_default())
    }

    /// What `namespace/repo` stores — packs and LFS objects — or `None`
    /// when it could not be measured.
    fn repo_stored(&self, namespace: &str, repo: &str) -> Option<u64> {
        let rows = self.repo_usage(namespace).ok()?;
        rows.into_iter()
            .find(|row| row.repo.as_ref().is_some_and(|p| p.repo == repo))
            .map(|row| row.stored_bytes.saturating_add(row.lfs_bytes))
    }

    /// **Move `from` to `to`**, everything it holds, through the write paths
    /// every push takes: a new repository at `to` made as `from` is (its hash,
    /// default branch, visibility, description), its objects and references
    /// copied in as ONE pack and ONE ref batch, HEAD, its LFS objects, its
    /// branch protections and ref-name policy — then `from` purged. Every step
    /// is a frame the transaction log already carries, so a twin and a restore
    /// see a move as a create, a push and a purge; no new kind of frame.
    ///
    /// A failure before the purge undoes the new repository and leaves `from`
    /// exactly as it was. The pack is held in memory once.
    pub(crate) fn move_repository(
        &self,
        from: &gunnar_proto::RepoPath,
        to: &gunnar_proto::RepoPath,
        by: &str,
    ) -> ControlResult<gunnar_proto::Repo> {
        let src = self.get_repo(from)?;
        if self.stores.catalog().namespace(&to.namespace).map_err(internal)?.is_none() {
            return Err(ControlError::NotFound(format!("account {}", to.namespace)));
        }
        if self.get_repo(to).is_ok() {
            return Err(ControlError::AlreadyExists(format!(
                "{}/{} already exists, so {}/{} cannot move there",
                to.namespace, to.repo, from.namespace, from.repo
            )));
        }
        self.create_repo(gunnar_grpc::NewRepo {
            path: to.clone(),
            hash_kind: gunnar_proto::HashKind::try_from(src.hash_kind).unwrap_or(gunnar_proto::HashKind::Sha256),
            default_branch: src.default_branch.clone(),
            visibility: gunnar_proto::Visibility::try_from(src.visibility).unwrap_or(gunnar_proto::Visibility::Private),
            description: src.description.clone(),
            adopt: false,
        })?;
        if let Err(err) = self.copy_repository(from, to) {
            if let Err(undo) = self.delete_repo(
                to,
                DeleteOptions {
                    purge: true,
                    retention_days: None,
                    principal_id: by.to_owned(),
                },
            ) {
                tracing::error!(
                    target: "gunnar::audit",
                    event = "repo.move_undo_failed",
                    to = %format!("{}/{}", to.namespace, to.repo),
                    reason = %undo,
                    "a failed move's new repository could not be purged again"
                );
            }
            return Err(ControlError::Internal(format!(
                "moving {}/{} to {}/{}: {err:#}; nothing was moved",
                from.namespace, from.repo, to.namespace, to.repo
            )));
        }
        // **Encrypted stays encrypted** (wiring tracer B 2026-10-08): set
        // after the copy (the copy is the members' own history, older
        // plaintext included, and is not re-judged) and before the purge, so a
        // move never leaves a repository that takes plaintext where the
        // members were promised it would not.
        if src.encrypted {
            if let Err(err) = self.set_encrypted(to, true) {
                let _ = self.delete_repo(
                    to,
                    DeleteOptions {
                        purge: true,
                        retention_days: None,
                        principal_id: by.to_owned(),
                    },
                );
                return Err(ControlError::Internal(format!(
                    "moving {}/{} to {}/{}: its encryption could not be carried ({err}); nothing was moved",
                    from.namespace, from.repo, to.namespace, to.repo
                )));
            }
        }
        // The policy goes with it: a protected branch stays protected.
        for rule in self.get_protection(from, None).unwrap_or_default() {
            if let Err(err) = self.set_protection(to, rule) {
                tracing::warn!(target: "gunnar::audit", event = "repo.move_policy_not_carried", reason = %err, "a branch protection did not move with its repository");
            }
        }
        if let Ok(policy) = self.get_ref_name_policy(from) {
            if policy != gunnar_proto::RefNamePolicy::default() {
                if let Err(err) = self.set_ref_name_policy(to, policy) {
                    tracing::warn!(target: "gunnar::audit", event = "repo.move_policy_not_carried", reason = %err, "a ref-name policy did not move with its repository");
                }
            }
        }
        // …and the old one goes for good: its bytes leave the clan's room.
        self.delete_repo(
            from,
            DeleteOptions {
                purge: true,
                retention_days: None,
                principal_id: by.to_owned(),
            },
        )?;
        self.get_repo(to)
    }

    /// The copy half of [`Self::move_repository`]: the snapshot's export of
    /// one repository (`twin::snapshot`), written into `to` through its
    /// logged store instead of into a stream.
    fn copy_repository(&self, from: &gunnar_proto::RepoPath, to: &gunnar_proto::RepoPath) -> anyhow::Result<()> {
        let from_ref = gunnar_core::RepoRef::new(&from.namespace, &from.repo)?;
        let to_ref = gunnar_core::RepoRef::new(&to.namespace, &to.repo)?;
        let src_row = self.stores.catalog().get(&from_ref)?.context("the repository to move is not in the catalog")?;
        let dst_row = self.stores.catalog().get(&to_ref)?.context("the new repository is not in the catalog")?;
        let src = self.stores.repo_store(&src_row)?;
        let dst = self.stores.repo_store(&dst_row)?;
        let git = src.unlogged();
        let mut updates = Vec::new();
        let mut tips: Vec<Vec<u8>> = Vec::new();
        for r in git.refs()? {
            if r.name == "HEAD" {
                continue;
            }
            if let Some(oid) = &r.oid {
                tips.push(oid.clone());
            }
            updates.push(RefUpdate {
                name: r.name,
                target: r.oid.as_deref().map(hex::encode),
                peeled: r.peeled.as_deref().map(hex::encode),
                symref_target: r.symref_target,
            });
        }
        if !tips.is_empty() {
            let want: Vec<&[u8]> = tips.iter().map(Vec::as_slice).collect();
            let objects = match git.select(&want, &[])? {
                Some(set) => set.objects,
                None => git.reachable(&want, &[])?,
            };
            let objs: Vec<&[u8]> = objects.iter().map(Vec::as_slice).collect();
            let mut pack = Vec::new();
            git.emit_pack(&objs, &[], &Caps::modern(), &mut pack)
                .context("exporting the objects to move")?;
            dst.serve().put_pack(&pack).context("writing the moved objects")?;
        }
        if !updates.is_empty() {
            dst.serve().put_refs(&updates).context("writing the moved references")?;
        }
        if let Some(target) = git.head()?.and_then(|h| h.symref_target) {
            dst.serve().set_head(&target).context("pointing the moved HEAD")?;
        }
        let lfs_dir = src.git_dir().join("lfs").join("objects");
        let files = lfs_files(&lfs_dir)?;
        if !files.is_empty() {
            let store = crate::txlog_lfs::LoggedLfs::new(
                gunnar_lfs::FsLfsStore::open_or_init(dst.git_dir())?,
                Arc::clone(self.stores.txlog()),
                format!("{}/{}", to.namespace, to.repo),
            );
            for path in files {
                let Some(oid) = path.file_name().and_then(|n| n.to_str()).and_then(|n| gunnar_lfs::Oid::parse(n).ok()) else {
                    continue;
                };
                let mut file = std::fs::File::open(&path).with_context(|| format!("opening LFS {}", path.display()))?;
                let size = file.metadata()?.len();
                let mut pending = store.begin_write(&oid, size)?;
                std::io::copy(&mut file, &mut pending).with_context(|| format!("copying LFS {}", path.display()))?;
                pending.commit()?;
            }
        }
        drop(src);
        drop(dst);
        self.stores.close_repo(&from_ref);
        self.stores.close_repo(&to_ref);
        Ok(())
    }
}

impl ClanPlane for CoreControlPlane {
    fn clan_create(&self, name: &str, creator: &str, contribution: u64, room: ClanRoom) -> ControlResult<gunnar_proto::Clan> {
        if name.contains('/') {
            return Err(ControlError::InvalidArgument(format!(
                "{name:?}: a clan's name is one namespace, with no `/` in it"
            )));
        }
        // The namespace first — the catalog's claim on the name.
        match self.stores.catalog().create_namespace(name) {
            Ok(_) => {}
            Err(gunnar_core::Error::NamespaceExists(_)) => {
                return Err(ControlError::AlreadyExists(format!("a namespace called {name}")))
            }
            Err(err) => return Err(internal(err)),
        }
        match self
            .stores
            .auth()
            .create_clan(name, creator, contribution, room_in(room))
        {
            Ok(clan) => Ok(self.clan_written(clan)),
            Err(err) => {
                // Made for this clan a moment ago, and empty: given back, so a
                // refused clan leaves no name taken.
                if let Err(undo) = self.stores.catalog().delete_namespace(name, false) {
                    tracing::warn!(
                        target: "gunnar::audit",
                        event = "clan.namespace_not_given_back",
                        clan = %name,
                        reason = %undo,
                        "a refused clan's namespace could not be removed again"
                    );
                }
                Err(clan_error(err))
            }
        }
    }

    fn clan_get(&self, name: &str) -> ControlResult<Option<gunnar_proto::Clan>> {
        Ok(self.stores.auth().clan(name).map_err(internal)?.as_ref().map(|c| self.clan_view(c)))
    }

    fn clans_of(&self, principal: &str) -> ControlResult<Vec<gunnar_proto::Clan>> {
        Ok(self
            .stores
            .auth()
            .clans_of(principal)
            .map_err(internal)?
            .iter()
            .map(|c| self.clan_view(c))
            .collect())
    }

    fn clan_invite(&self, clan: &str, by: &str, principal: &str) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .invite_to_clan(clan, by, principal)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_answer(
        &self,
        clan: &str,
        principal: &str,
        accept: bool,
        contribution: u64,
        room: ClanRoom,
    ) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .answer_clan_invite(clan, principal, accept, contribution, room_in(room))
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_withdraw(&self, clan: &str, by: &str, principal: &str) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .withdraw_clan_invite(clan, by, principal)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_set_contribution(
        &self,
        clan: &str,
        member: &str,
        bytes: u64,
        room: ClanRoom,
        used: Option<u64>,
    ) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .set_clan_contribution(clan, member, bytes, room_in(room), used, &self.lapsed_in(clan)?)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_leave(&self, clan: &str, member: &str, used: Option<u64>) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .leave_clan(clan, member, used, &self.lapsed_in(clan)?)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_remove(&self, clan: &str, by: &str, member: &str, used: Option<u64>) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .remove_from_clan(clan, by, member, used, &self.lapsed_in(clan)?)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_set_admin(&self, clan: &str, by: &str, member: &str, admin: bool) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .set_clan_admin(clan, by, member, admin)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    /// The admin's, and only of an EMPTY clan: no repository in its
    /// namespace, live or deleted-and-still-held. The namespace goes first —
    /// the catalog refuses a namespace that holds a repository, in its own
    /// write, so a repository made a moment ago keeps the clan — and the clan
    /// row after it.
    fn clan_dissolve(&self, clan: &str, by: &str) -> ControlResult<gunnar_proto::Clan> {
        let auth = self.stores.auth();
        auth.clan_for(clan, by).map_err(clan_error)?;
        let live = self.stores.catalog().list(Some(clan)).map_err(internal)?.len();
        let held = gunnar_grpc::ControlPlane::list_deleted(self, Some(clan))?.len();
        if live + held > 0 {
            return Err(ControlError::FailedPrecondition(format!(
                "clan {clan} still holds {live} repositor{} and {held} deleted one{} kept for their \
                 window: its admin deletes and purges them before the clan can be dissolved",
                if live == 1 { "y" } else { "ies" },
                if held == 1 { "" } else { "s" },
            )));
        }
        if self.stores.catalog().namespace(clan).map_err(internal)?.is_some() {
            self.stores
                .catalog()
                .delete_namespace(clan, false)
                .map_err(internal)?;
            if let Err(err) = self.stores.flow().forget(clan) {
                tracing::warn!(
                    target: "gunnar::store",
                    event = "flow.forget_failed",
                    namespace = %clan,
                    %err,
                    "the clan's namespace was deleted but its activity series could not be removed"
                );
            }
        }
        let row = auth.dissolve_clan(clan, by).map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_room_left(&self, account: &str, room: ClanRoom) -> ControlResult<Option<u64>> {
        self.stores
            .auth()
            .room_left(account, room_in(room))
            .map_err(internal)
    }

    fn clan_given(&self, account: &str) -> ControlResult<u64> {
        self.stores.auth().contributed_by(account).map_err(internal)
    }

    fn clan_offer_repo(&self, clan: &str, by: &str, repo: &str, to: &str) -> ControlResult<gunnar_proto::Clan> {
        // The clan's admin first — an outsider learns nothing about the
        // clan's repositories — then the repository must be the clan's.
        self.stores.auth().clan_for(clan, by).map_err(clan_error)?;
        self.get_repo(&gunnar_proto::RepoPath::new(clan, repo))?;
        let row = self
            .stores
            .auth()
            .offer_clan_repo(clan, by, repo, to)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_repo_offer(&self, clan: &str, repo: &str) -> ControlResult<Option<gunnar_proto::RepoOffer>> {
        Ok(self
            .stores
            .auth()
            .clan_repo_offer(clan, repo)
            .map_err(internal)?
            .map(|offer| offer_out(clan, &offer, self.repo_stored(clan, repo))))
    }

    fn clan_offers_to(&self, to: &str) -> ControlResult<Vec<gunnar_proto::RepoOffer>> {
        Ok(self
            .stores
            .auth()
            .clan_offers_to(to)
            .map_err(internal)?
            .iter()
            .map(|(clan, offer)| offer_out(clan, offer, self.repo_stored(clan, &offer.repo)))
            .collect())
    }

    fn clan_withdraw_offer(&self, clan: &str, by: &str, repo: &str) -> ControlResult<gunnar_proto::Clan> {
        let row = self
            .stores
            .auth()
            .withdraw_clan_repo_offer(clan, by, repo)
            .map_err(clan_error)?;
        Ok(self.clan_written(row))
    }

    fn clan_decline_offer(&self, clan: &str, repo: &str, to: &str) -> ControlResult<gunnar_proto::RepoOffer> {
        let offer = self
            .stores
            .auth()
            .close_clan_repo_offer(clan, repo, to)
            .map_err(clan_error)?;
        self.stores.mirror_identity();
        Ok(offer_out(clan, &offer, None))
    }

    /// The move, then the offer closed. A move that fails leaves the offer
    /// open (and the repository where it was) for another try or a decline.
    fn clan_take_offer(
        &self,
        clan: &str,
        repo: &str,
        to: &str,
        by: &str,
    ) -> ControlResult<(gunnar_proto::RepoOffer, gunnar_proto::Repo)> {
        let stored = self.repo_stored(clan, repo);
        let open = self
            .stores
            .auth()
            .clan_repo_offer(clan, repo)
            .map_err(internal)?
            .filter(|o| o.to == to)
            .ok_or_else(|| ControlError::NotFound(format!("an offer of {clan}/{repo} to {to}")))?;
        let moved = self.move_repository(
            &gunnar_proto::RepoPath::new(clan, repo),
            &gunnar_proto::RepoPath::new(to, repo),
            by,
        )?;
        let offer = self
            .stores
            .auth()
            .close_clan_repo_offer(clan, repo, to)
            .unwrap_or(open);
        self.stores.mirror_identity();
        Ok((offer_out(clan, &offer, stored), moved))
    }

    fn clan_role(&self, principal: &str, namespace: &str) -> ControlResult<Option<gunnar_proto::ClanRole>> {
        Ok(self
            .stores
            .auth()
            .clan_role(principal, namespace)
            .map_err(internal)?
            .map(|role| match role {
                ClanRole::Admin => gunnar_proto::ClanRole::Admin,
                ClanRole::Member => gunnar_proto::ClanRole::Member,
                ClanRole::Invited => gunnar_proto::ClanRole::Invited,
            }))
    }
}

/// Every LFS object file under `dir` (the snapshot's walk, `twin::snapshot`).
fn lfs_files(dir: &std::path::Path) -> anyhow::Result<Vec<std::path::PathBuf>> {
    let mut out = Vec::new();
    let mut stack = vec![dir.to_path_buf()];
    while let Some(d) = stack.pop() {
        let entries = match std::fs::read_dir(&d) {
            Ok(entries) => entries,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
            Err(e) => return Err(e).with_context(|| format!("listing {}", d.display())),
        };
        for entry in entries {
            let entry = entry?;
            let kind = entry.file_type()?;
            if kind.is_dir() {
                stack.push(entry.path());
            } else if kind.is_file() {
                out.push(entry.path());
            }
        }
    }
    out.sort();
    Ok(out)
}