use semver::Version;
use crate::CodePushError;
use crate::layout::ChannelIndex;
use crate::layout::INDEX_FORMAT_VERSION;
use crate::layout::ObjectKey;
use crate::layout::ObjectKind;
use crate::layout::StaticObject;
use crate::layout::release_name;
use crate::manifest::BundleManifest;
use crate::manifest::container_manifest;
use crate::provenance::ReleaseApprovals;
use crate::signing::TrustedKeys;
use crate::store::ReleaseRecord;
use crate::store::ReleaseStore;
pub trait Bucket {
fn get(&mut self, key: &ObjectKey) -> Result<Option<Vec<u8>>, CodePushError>;
fn put(&mut self, object: &StaticObject) -> Result<(), CodePushError>;
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublishReport {
pub version: Version,
pub build: u64,
pub releases: Vec<String>,
pub approvals: Vec<String>,
pub written: usize,
pub already_present: usize,
}
pub fn publish_channel(
store: &ReleaseStore,
app: &str,
channel: &str,
bucket: &mut impl Bucket,
) -> Result<PublishReport, CodePushError> {
let records = store.list(app, channel)?;
let latest = records
.first()
.ok_or_else(|| CodePushError::NoReleases(format!("{app}/{channel}")))?;
let latest_name = release_name(&latest.manifest.version, latest.manifest.build);
if latest.signature.is_none() {
return Err(CodePushError::Unsigned(format!(
"{app}/{channel}/{latest_name}"
)));
}
let index_key = ObjectKey::index(app, channel);
let remote = bucket
.get(&index_key)?
.map(|bytes| parse_index(&bytes))
.transpose()?;
let remote_id = remote
.as_ref()
.map(|index| (&index.manifest.version, index.manifest.build));
if let (Some(remote), Some(remote_id)) = (&remote, remote_id) {
let remote_name = release_name(remote_id.0, remote_id.1);
if remote_id > latest.id() {
return Err(CodePushError::RegistryAhead {
registry: remote_name,
store: latest_name,
});
}
let diverged = records.iter().any(|record| {
record.id() == remote_id
&& record.manifest.bundle_hash() != remote.manifest.bundle_hash()
});
if diverged {
return Err(CodePushError::RegistryConflict(format!(
"{app}/{channel}/{remote_name}"
)));
}
}
let mut report = PublishReport {
version: latest.manifest.version.clone(),
build: latest.manifest.build,
releases: Vec::new(),
approvals: Vec::new(),
written: 0,
already_present: 0,
};
let pending = records
.iter()
.rev()
.filter(|record| remote_id.is_none_or(|remote_id| record.id() > remote_id));
for record in pending {
for object in store.release_objects(record)? {
if object.key.kind() == ObjectKind::Delta
&& object.bytes.len() as u64 > ObjectKind::Delta.max_bytes()
{
continue;
}
match bucket.get(&object.key)? {
Some(existing) if existing == object.bytes => report.already_present += 1,
Some(_) => {
return Err(CodePushError::RegistryConflict(
object.key.as_str().to_owned(),
));
}
None => {
bucket.put(&object)?;
report.written += 1;
}
}
}
report.releases.push(release_name(
&record.manifest.version,
record.manifest.build,
));
}
publish_approvals(store, &records, bucket, &mut report)?;
let index = latest.channel_index();
if remote.as_ref() != Some(&index) {
bucket.put(&StaticObject {
key: index_key,
bytes: serde_json::to_vec_pretty(&index)?,
})?;
report.written += 1;
}
Ok(report)
}
fn publish_approvals(
store: &ReleaseStore,
records: &[ReleaseRecord],
bucket: &mut impl Bucket,
report: &mut PublishReport,
) -> Result<(), CodePushError> {
for record in records.iter().rev() {
let Some(local) = store.approvals(&record.manifest)? else {
continue;
};
if merge_approvals(bucket, &record.manifest, local)? {
report.written += 1;
report.approvals.push(release_name(
&record.manifest.version,
record.manifest.build,
));
} else {
report.already_present += 1;
}
}
Ok(())
}
fn merge_approvals(
bucket: &mut impl Bucket,
manifest: &BundleManifest,
local: ReleaseApprovals,
) -> Result<bool, CodePushError> {
let key = ObjectKey::approvals_of(manifest);
let conflict = || CodePushError::RegistryConflict(key.as_str().to_owned());
let remote = match bucket.get(&key)? {
Some(bytes) => Some(ReleaseApprovals::from_slice(&bytes).map_err(|_| conflict())?),
None => None,
};
let present = remote.is_some();
let mut merged = remote.unwrap_or_else(ReleaseApprovals::empty);
if merged.merge(local.approvals)? == 0 && present {
return Ok(false);
}
bucket.put(&StaticObject {
key,
bytes: merged.to_vec()?,
})?;
Ok(true)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PullReport {
Empty,
Pulled {
version: Version,
build: u64,
imported: bool,
},
}
pub fn pull_channel(
store: &ReleaseStore,
app: &str,
channel: &str,
bucket: &mut impl Bucket,
trusted: &TrustedKeys,
) -> Result<PullReport, CodePushError> {
let Some(index_bytes) = bucket.get(&ObjectKey::index(app, channel))? else {
return Ok(PullReport::Empty);
};
let index = parse_index(&index_bytes)?;
if index.format_version != INDEX_FORMAT_VERSION {
return Err(CodePushError::UnsupportedFormatVersion(
index.format_version,
));
}
let manifest = &index.manifest;
if manifest.app != app || manifest.channel != channel {
return Err(CodePushError::WrongTarget {
tracked: format!("{app}/{channel}"),
offered: format!("{}/{}", manifest.app, manifest.channel),
});
}
let signature = index.signature.as_ref().ok_or_else(|| {
CodePushError::Unsigned(format!(
"{app}/{channel}/{}",
release_name(&manifest.version, manifest.build)
))
})?;
trusted.verify(manifest, Some(signature))?;
let (version, build) = (&manifest.version, manifest.build);
let release_key = ObjectKey::release(app, channel, version, build);
let container = bucket
.get(&release_key)?
.ok_or_else(|| CodePushError::NotFound(release_key.as_str().to_owned()))?;
let (container_manifest, _) = container_manifest(&container)?;
if container_manifest.bundle_hash() != manifest.bundle_hash() {
return Err(CodePushError::HashMismatch {
path: release_key.as_str().to_owned(),
expected: manifest.bundle_hash(),
actual: container_manifest.bundle_hash(),
});
}
let imported = match store.import_release(&container, signature) {
Ok(imported) => imported,
Err(CodePushError::ReleaseExists(release)) => {
return Err(CodePushError::RegistryConflict(release));
}
Err(error) => return Err(error),
};
Ok(PullReport::Pulled {
version: version.clone(),
build,
imported,
})
}
fn parse_index(bytes: &[u8]) -> Result<ChannelIndex, CodePushError> {
if bytes.len() as u64 > ObjectKind::Index.max_bytes() {
return Err(CodePushError::Malformed("index exceeds 2 MiB".to_owned()));
}
Ok(serde_json::from_slice(bytes)?)
}