use std::collections::{HashMap, HashSet, VecDeque};
use crate::catalog_types::{
MAX_NAMESPACE_DEPTH, NamespaceChange, NamespaceParent, NamespacePath, NamespaceSegment,
PageSize, ProjectResourceState, ReadJournalStateRequest, ResourceAtSnapshot, ResourceChange,
ResourceId,
};
use crate::{Error, ResourceAddress, Snapshot};
pub(crate) struct Binding {
pub parent: NamespaceParent,
pub name: NamespaceSegment,
pub resource: ResourceAtSnapshot,
}
pub(crate) struct Namespace {
resources: HashMap<ResourceId, ProjectResourceState>,
bindings: HashMap<(NamespaceParent, NamespaceSegment), ResourceId>,
}
pub(crate) fn resource_path(path: &str) -> Result<NamespacePath, Error> {
let address: ResourceAddress = path.parse()?;
if address.reference().is_some() {
return Err(Error::invalid(
"use a snapshot to select the Connection's branch or commit",
));
}
Ok(address.path().clone())
}
impl Namespace {
pub async fn read(snapshot: &Snapshot) -> Result<Self, Error> {
let mut state = Self {
resources: HashMap::new(),
bindings: HashMap::new(),
};
let mut page_token = None;
let mut seen = HashSet::new();
loop {
let page = snapshot
.project
.client
.catalog()
.read_journal_state(
&snapshot.project.id,
snapshot.commit_id,
&ReadJournalStateRequest {
page_token,
page_size: Some(PageSize::default()),
},
)
.await?;
if page.commit_id != snapshot.commit_id {
return Err(Error::Protocol(
"Catalog changed the selected commit".into(),
));
}
for resource in page.resources {
resource.validate().map_err(Error::invalid)?;
state
.resources
.insert(resource.descriptor.resource_id, resource);
}
for entry in page.namespace {
entry.parent.validate().map_err(Error::invalid)?;
entry.name.validate().map_err(Error::invalid)?;
entry.resource.validate().map_err(Error::invalid)?;
let id = entry.resource.descriptor.resource_id;
state.resources.insert(id, entry.resource);
state.bindings.insert((entry.parent, entry.name), id);
}
let Some(next) = page.next_page_token else {
return Ok(state);
};
next.validate().map_err(Error::invalid)?;
if !seen.insert(next.clone()) {
return Err(Error::Protocol("Catalog repeated a page token".into()));
}
page_token = Some(next);
}
}
pub fn apply(
&mut self,
resources: impl IntoIterator<Item = ResourceChange>,
namespace: impl IntoIterator<Item = NamespaceChange>,
) -> Result<(), Error> {
for change in resources {
match change {
ResourceChange::Create(change) => {
self.resources.insert(
change.descriptor.resource_id,
ProjectResourceState {
descriptor: change.descriptor,
version_id: change.version_id,
},
);
}
ResourceChange::Advance(change) => {
if let Some(resource) = self.resources.get_mut(&change.resource_id) {
resource.version_id = change.replacement;
}
}
ResourceChange::Retire(change) => {
self.resources.remove(&change.resource_id);
}
ResourceChange::Assert(_) => {}
ResourceChange::Unknown { .. } => {
return Err(Error::invalid("unknown resource change"));
}
}
}
for change in namespace {
match change {
NamespaceChange::Bind(change) => {
self.bindings
.insert((change.parent, change.name), change.replacement);
}
NamespaceChange::Unbind(change) => {
self.bindings.remove(&(change.parent, change.name));
}
NamespaceChange::Assert(_) => {}
NamespaceChange::Unknown { .. } => {
return Err(Error::invalid("unknown namespace change"));
}
}
}
Ok(())
}
pub fn resolve(&self, path: &NamespacePath) -> Result<Option<Binding>, Error> {
let parts = path.segments();
let mut pending = VecDeque::from([(NamespaceParent::project_root(), 0)]);
let mut visited = HashSet::new();
while let Some((anchor, start)) = pending.pop_front() {
if !visited.insert((anchor.clone(), start)) {
continue;
}
for end in ((start + 1)..=parts.len()).rev() {
let parent = prefix_parent(anchor.clone(), &parts[start..end - 1])?;
let name = &parts[end - 1];
let Some(id) = self.bindings.get(&(parent.clone(), name.clone())) else {
continue;
};
let Some(resource) = self.resources.get(id) else {
continue;
};
if end == parts.len() {
return Ok(Some(Binding {
parent,
name: name.clone(),
resource: ResourceAtSnapshot {
descriptor: resource.descriptor.clone(),
version_id: resource.version_id,
},
}));
}
pending.push_back((NamespaceParent::resource(*id), end));
}
}
Ok(None)
}
pub fn parent(
&self,
path: &NamespacePath,
) -> Result<(NamespaceParent, NamespaceSegment), Error> {
if let Some(binding) = self.resolve(path)? {
return Ok((binding.parent, binding.name));
}
let (name, prefix) = path.segments().split_last().expect("paths are nonempty");
for end in (1..=prefix.len()).rev() {
let ancestor = NamespacePath::new(prefix[..end].to_vec()).map_err(Error::invalid)?;
if let Some(binding) = self.resolve(&ancestor)? {
return Ok((
prefix_parent(
NamespaceParent::resource(binding.resource.descriptor.resource_id),
&prefix[end..],
)?,
name.clone(),
));
}
}
Ok((
prefix_parent(NamespaceParent::project_root(), prefix)?,
name.clone(),
))
}
pub fn named(&self, kind: &str) -> Result<Vec<(String, ResourceAtSnapshot)>, Error> {
let mut by_anchor = HashMap::<Option<ResourceId>, Vec<_>>::new();
for ((parent, name), id) in &self.bindings {
by_anchor
.entry(parent.resource_id().copied())
.or_default()
.push((parent, name, id));
}
let mut pending = VecDeque::from([(None, Vec::new())]);
let mut named = Vec::new();
while let Some((anchor, prefix)) = pending.pop_front() {
for (parent, name, id) in by_anchor.get(&anchor).into_iter().flatten() {
let Some(resource) = self.resources.get(id) else {
continue;
};
let mut path = prefix.clone();
path.extend(parent.path_segments().iter().cloned());
path.push((*name).clone());
if path.len() > MAX_NAMESPACE_DEPTH {
return Err(Error::Protocol(
"Catalog namespace exceeds the maximum path depth".into(),
));
}
if resource.descriptor.kind.0 == kind {
named.push((
path.iter()
.map(NamespaceSegment::as_str)
.collect::<Vec<_>>()
.join("/"),
ResourceAtSnapshot {
descriptor: resource.descriptor.clone(),
version_id: resource.version_id,
},
));
}
pending.push_back((Some(**id), path));
}
}
named.sort_by(|left, right| left.0.cmp(&right.0));
Ok(named)
}
}
fn prefix_parent(
parent: NamespaceParent,
parts: &[NamespaceSegment],
) -> Result<NamespaceParent, Error> {
parts.iter().try_fold(parent, |parent, part| {
parent.child_prefix(part.clone()).map_err(Error::invalid)
})
}