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};
#[derive(Clone)]
pub struct Project {
pub(crate) client: Client,
pub(crate) id: ProjectId,
}
impl Client {
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 {
pub fn client(&self) -> &Client {
&self.client
}
pub fn id(&self) -> &ProjectId {
&self.id
}
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,
))
}
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,
))
}
pub fn snapshot_at(&self, commit_id: ProjectCommitId) -> Snapshot {
Snapshot {
project: self.clone(),
commit_id,
branch_id: None,
}
}
pub fn resume_commit(&self, saved: SavedCommit) -> Result<FrozenCommit, Error> {
FrozenCommit::resume(self, saved)
}
#[cfg(feature = "tables")]
pub async fn table(&self, branch: &str, path: &str) -> Result<Table, Error> {
self.table_session(branch).await?.table(path).await
}
}
#[derive(Clone)]
pub struct Snapshot {
pub(crate) project: Project,
pub(crate) commit_id: ProjectCommitId,
pub(crate) branch_id: Option<ProjectBranchId>,
}
impl Snapshot {
pub fn commit_id(&self) -> ProjectCommitId {
self.commit_id
}
pub fn branch_id(&self) -> Option<ProjectBranchId> {
self.branch_id
}
pub fn project(&self) -> &Project {
&self.project
}
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);
}
}
#[cfg(feature = "tables")]
pub async fn table(&self, path: &str) -> Result<Table, Error> {
self.table_session().table(path).await
}
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)
}
}
#[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,
}
}
#[cfg(feature = "tables")]
pub(crate) fn of(snapshot: &Snapshot) -> Option<Self> {
Some(Self {
snapshot: snapshot.clone(),
branch_id: snapshot.branch_id?,
})
}
pub fn branch_id(&self) -> ProjectBranchId {
self.branch_id
}
pub fn snapshot(&self) -> &Snapshot {
&self.snapshot
}
pub fn commit(&self) -> Commit {
Commit::new(self.clone())
}
}
impl Deref for BranchHead {
type Target = Snapshot;
fn deref(&self) -> &Snapshot {
&self.snapshot
}
}
impl Project {
pub async fn get(&self) -> Result<ProjectInfo, Error> {
Ok(self
.client
.projects()
.read_project(self.id.as_str().into())
.await?)
}
pub async fn delete(&self) -> Result<(), Error> {
Ok(self
.client
.projects()
.delete_project(self.id.as_str().into())
.await?)
}
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,
},
)
}
pub async fn grant_role(&self, request: GrantRoleRequest) -> Result<GrantRoleResponse, Error> {
Ok(self
.client
.authorization()
.grant_role(self.id.as_str().into(), request)
.await?)
}
pub async fn presentation(&self, key: PresentationKey) -> Result<Presentation, Error> {
Ok(self
.client
.projects()
.read_presentation(self.id.as_str().into(), key)
.await?)
}
pub async fn save_presentation(
&self,
request: SavePresentation,
) -> Result<Presentation, Error> {
Ok(self
.client
.projects()
.save_presentation(self.id.as_str().into(), request)
.await?)
}
}