spiral/

project.rs

//! Project handles, immutable Catalog commit snapshots and pinned branch heads.
use std::collections::HashSet;
use std::ops::Deref;

#[cfg(feature = "tables")]
use crate::Table;
use crate::api::authorization::ListGrantsRequest;
use crate::catalog_types::{
    CommitSelector, CommitTarget, CreateBranchRequest, IdempotencyKey, NamespacePath,
    NamespaceSegment, PageSize, ProjectBranchId, ProjectBranchName, ProjectCommitId, ProjectId,
    ProjectResourceState, ReadJournalStateRequest, ResolveJournalPathRequest, ResourceAtSnapshot,
};
use crate::commit::{Commit, FrozenCommit, SavedCommit};
use crate::resources::project::{
    Grant, GrantRoleRequest, GrantRoleResponse, Presentation, PresentationKey, Principal,
    Project as ProjectInfo, SavePresentation,
};
#[cfg(feature = "tables")]
use crate::table::CatalogBase;
use crate::{Client, Error, List};

/// Entry point for project resources, bound to one authenticated connection.
#[derive(Clone)]
pub struct Project {
    pub(crate) client: Client,
    pub(crate) id: ProjectId,
}
impl Client {
    /// Select a project without making a network request.
    pub fn project(&self, id: impl AsRef<str>) -> Result<Project, Error> {
        Ok(Project {
            client: self.clone(),
            id: ProjectId::new(id.as_ref()).map_err(Error::invalid)?,
        })
    }
}
impl Project {
    /// The client whose credentials and execution backend this Project uses.
    pub fn client(&self) -> &Client {
        &self.client
    }

    /// Stable project identifier.
    pub fn id(&self) -> &ProjectId {
        &self.id
    }
    /// Pin the current head of a branch, which a [`Commit`] can publish onto.
    pub async fn branch(&self, name: &str) -> Result<BranchHead, Error> {
        let branch = ProjectBranchName::new(name).map_err(Error::invalid)?;
        let pinned = self
            .client
            .connection
            .catalog
            .pin_journal(&self.id, &branch)
            .await?;
        pinned.validate().map_err(Error::invalid)?;
        Ok(BranchHead::new(
            self.clone(),
            pinned.head_commit_id,
            pinned.branch_id,
        ))
    }
    /// Create a branch at an existing project commit and pin its head.
    pub async fn create_branch(
        &self,
        name: &str,
        commit_id: ProjectCommitId,
    ) -> Result<BranchHead, Error> {
        ProjectCommitId::new(commit_id.0).map_err(Error::invalid)?;
        let branch = self
            .client
            .connection
            .catalog
            .create_journal_branch(
                &self.id,
                &CreateBranchRequest {
                    name: ProjectBranchName::new(name).map_err(Error::invalid)?,
                    target: CommitSelector::Commit(CommitTarget { commit_id }),
                    idempotency_key: IdempotencyKey::new(format!("branch-{name}-{commit_id}"))
                        .map_err(Error::invalid)?,
                },
            )
            .await?;
        branch.validate().map_err(Error::invalid)?;
        Ok(BranchHead::new(
            self.clone(),
            branch.head_commit_id,
            branch.branch_id,
        ))
    }

    /// Open an exact commit. Its snapshot is read-only: commits publish onto a [`BranchHead`].
    pub fn snapshot_at(&self, commit_id: ProjectCommitId) -> Snapshot {
        Snapshot {
            project: self.clone(),
            commit_id,
            branch_id: None,
        }
    }
    /// Resume a commit saved with [`FrozenCommit::save`], against its original base and branch.
    pub fn resume_commit(&self, saved: SavedCommit) -> Result<FrozenCommit, Error> {
        FrozenCommit::resume(self, saved)
    }
    /// Pin the branch head and resolve a table to an immutable version.
    #[cfg(feature = "tables")]
    pub async fn table(&self, branch: &str, path: &str) -> Result<Table, Error> {
        self.table_session(branch).await?.table(path).await
    }
}

/// Immutable project state selected by one exact Catalog commit.
///
/// A snapshot pinned from a branch also records the branch; [`BranchHead`] is that pin.
#[derive(Clone)]
pub struct Snapshot {
    pub(crate) project: Project,
    pub(crate) commit_id: ProjectCommitId,
    pub(crate) branch_id: Option<ProjectBranchId>,
}
impl Snapshot {
    /// Exact commit selected by this snapshot.
    pub fn commit_id(&self) -> ProjectCommitId {
        self.commit_id
    }
    /// Stable branch selected when pinning, or None for an exact-commit snapshot.
    pub fn branch_id(&self) -> Option<ProjectBranchId> {
        self.branch_id
    }
    /// Project owning this state.
    pub fn project(&self) -> &Project {
        &self.project
    }
    /// Read resources at this commit, following bounded server pages.
    pub async fn resources(&self) -> Result<Vec<ProjectResourceState>, Error> {
        let mut resources = Vec::new();
        let mut page_token = None;
        let mut seen = HashSet::new();
        loop {
            let page = self
                .project
                .client
                .connection
                .catalog
                .read_journal_state(
                    &self.project.id,
                    self.commit_id,
                    &ReadJournalStateRequest {
                        page_token: page_token.clone(),
                        page_size: Some(PageSize::default()),
                    },
                )
                .await?;
            if page.commit_id != self.commit_id {
                return Err(Error::invalid("Catalog changed the selected commit"));
            }
            for resource in &page.resources {
                resource.validate().map_err(Error::invalid)?;
            }
            resources.extend(page.resources);
            let Some(next) = page.next_page_token else {
                return Ok(resources);
            };
            next.validate().map_err(Error::invalid)?;
            if !seen.insert(next.clone()) {
                return Err(Error::invalid("Catalog repeated a page token"));
            }
            page_token = Some(next);
        }
    }
    /// Resolve a resource address with this commit as its default snapshot.
    /// Use [`Self::table_session`] to share explicit branch pins across lookups.
    #[cfg(feature = "tables")]
    pub async fn table(&self, path: &str) -> Result<Table, Error> {
        self.table_session().table(path).await
    }

    /// Resolve the resource that `path` names at this commit.
    pub(crate) async fn resource_at_path(
        &self,
        path: &NamespacePath,
    ) -> Result<ResourceAtSnapshot, Error> {
        let resolved = self
            .project
            .client
            .connection
            .catalog
            .resolve_journal_path(
                &self.project.id,
                self.commit_id,
                &ResolveJournalPathRequest { path: path.clone() },
            )
            .await?;
        if resolved.commit_id != self.commit_id {
            return Err(Error::invalid("Catalog changed the selected commit"));
        }
        let resource = resolved.resource.ok_or_else(|| {
            Error::NotFound(
                path.segments()
                    .iter()
                    .map(NamespaceSegment::as_str)
                    .collect::<Vec<_>>()
                    .join("/"),
            )
        })?;
        resource.validate().map_err(Error::invalid)?;
        Ok(resource)
    }

    #[cfg(feature = "tables")]
    pub(crate) async fn table_at_path(&self, path: &NamespacePath) -> Result<Table, Error> {
        let resource = self.resource_at_path(path).await?;
        if resource.descriptor.kind.0 != "table" || resource.descriptor.service.as_str() != "arbor"
        {
            return Err(Error::invalid("resource is not an Arbor table"));
        }
        let mut table = self.project.client.table_by_id(
            resource.descriptor.resource_id.as_uuid(),
            resource.version_id.as_uuid(),
        );
        table.base = Some(CatalogBase {
            project_id: self.project.id.clone(),
            commit_id: self.commit_id,
            branch_id: self.branch_id,
            version_id: resource.version_id.as_uuid(),
            address: None,
        });
        Ok(table)
    }
}

/// The head of a branch, pinned when it was read: a snapshot that a [`Commit`] can publish onto.
///
/// The pin is fixed. Publishing returns the new head, and only succeeds while the branch
/// still points at this one.
#[derive(Clone)]
pub struct BranchHead {
    snapshot: Snapshot,
    branch_id: ProjectBranchId,
}
impl BranchHead {
    pub(crate) fn new(
        project: Project,
        commit_id: ProjectCommitId,
        branch_id: ProjectBranchId,
    ) -> Self {
        Self {
            snapshot: Snapshot {
                project,
                commit_id,
                branch_id: Some(branch_id),
            },
            branch_id,
        }
    }
    /// The head of the branch `snapshot` was pinned from, or `None` for an exact commit.
    #[cfg(feature = "tables")]
    pub(crate) fn of(snapshot: &Snapshot) -> Option<Self> {
        Some(Self {
            snapshot: snapshot.clone(),
            branch_id: snapshot.branch_id?,
        })
    }
    /// Stable identity of the branch, which survives renames.
    pub fn branch_id(&self) -> ProjectBranchId {
        self.branch_id
    }
    /// The pinned head as a read-only snapshot.
    pub fn snapshot(&self) -> &Snapshot {
        &self.snapshot
    }
    /// Start a commit onto this head.
    pub fn commit(&self) -> Commit {
        Commit::new(self.clone())
    }
}
impl Deref for BranchHead {
    type Target = Snapshot;
    fn deref(&self) -> &Snapshot {
        &self.snapshot
    }
}

impl Project {
    /// Read current project metadata.
    pub async fn get(&self) -> Result<ProjectInfo, Error> {
        Ok(self
            .client
            .projects()
            .read_project(self.id.as_str().into())
            .await?)
    }
    /// Delete this project. Requires project administration permission.
    pub async fn delete(&self) -> Result<(), Error> {
        Ok(self
            .client
            .projects()
            .delete_project(self.id.as_str().into())
            .await?)
    }
    /// List access grants, optionally restricted to one principal.
    pub fn grants(&self, principal: Option<Principal>) -> List<'_, Grant> {
        self.client.authorization().list_grants_list(
            self.id.as_str().into(),
            ListGrantsRequest {
                principal,
                page_size: None,
                page_token: None,
            },
        )
    }
    /// Grant a role to a principal on this project.
    pub async fn grant_role(&self, request: GrantRoleRequest) -> Result<GrantRoleResponse, Error> {
        Ok(self
            .client
            .authorization()
            .grant_role(self.id.as_str().into(), request)
            .await?)
    }
    /// Read shared presentation state at a key.
    pub async fn presentation(&self, key: PresentationKey) -> Result<Presentation, Error> {
        Ok(self
            .client
            .projects()
            .read_presentation(self.id.as_str().into(), key)
            .await?)
    }
    /// Save shared presentation state against its expected revision.
    pub async fn save_presentation(
        &self,
        request: SavePresentation,
    ) -> Result<Presentation, Error> {
        Ok(self
            .client
            .projects()
            .save_presentation(self.id.as_str().into(), request)
            .await?)
    }
}