use std::io;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::filesystem::ConfinedDir;
const STATE_DIR: &str = ".aion-authoring";
const REVISION_DIR: &str = "revisions";
const DEPLOYMENT_DIR: &str = "deployments";
const HEX: &[u8; 16] = b"0123456789abcdef";
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct Revision {
pub content_hash: String,
pub source: String,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct DeploymentRecord {
pub deployment_id: String,
pub document_path: String,
pub content_hash: String,
pub package_id: String,
pub workflow_type: String,
pub task_queue: String,
pub workflow_id: Option<String>,
pub run_id: Option<String>,
}
#[derive(Debug, thiserror::Error)]
pub enum RevisionError {
#[error("invalid content hash: {0}")]
InvalidHash(String),
#[error("document revision was not found: {0}")]
NotFound(String),
#[error("revision store I/O failed: {0}")]
Io(#[from] io::Error),
#[error("deployment record is invalid: {0}")]
InvalidRecord(String),
}
#[must_use]
pub fn content_hash(source: &str) -> String {
let mut digest = Sha256::new();
digest.update(source.as_bytes());
let bytes = digest.finalize();
let mut encoded = String::with_capacity(bytes.len() * 2);
for byte in bytes {
encoded.push(char::from(HEX[usize::from(byte >> 4)]));
encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
}
encoded
}
pub async fn store(root: &Path, source: &str) -> Result<Revision, RevisionError> {
let root = root.to_owned();
let source = source.to_owned();
blocking("revision store", move || {
let workspace = ConfinedDir::open_or_create(&root)?;
store_sync(&workspace, &source)
})
.await
}
pub async fn fetch(root: &Path, hash: &str) -> Result<Revision, RevisionError> {
validate_hash(hash)?;
let root = root.to_owned();
let hash = hash.to_owned();
blocking("revision fetch", move || {
let workspace = ConfinedDir::open(&root)?;
fetch_sync(&workspace, &hash)
})
.await
}
pub async fn record_deployment(
root: &Path,
record: &DeploymentRecord,
) -> Result<(), RevisionError> {
validate_identifier(&record.deployment_id, "deployment id")?;
validate_hash(&record.content_hash)?;
let root = root.to_owned();
let record = record.clone();
blocking("deployment record", move || {
let workspace = ConfinedDir::open_or_create(&root)?;
let revision = fetch_sync(&workspace, &record.content_hash)?;
if revision.content_hash != record.content_hash {
return Err(RevisionError::InvalidRecord(
"deployment revision identity changed".to_owned(),
));
}
let path = deployment_path(&record.deployment_id);
let bytes = serde_json::to_vec_pretty(&record)
.map_err(|error| RevisionError::InvalidRecord(error.to_string()))?;
workspace.atomic_write(&path, &bytes)?;
Ok(())
})
.await
}
pub async fn deployment(
root: &Path,
deployment_id: &str,
) -> Result<DeploymentRecord, RevisionError> {
validate_identifier(deployment_id, "deployment id")?;
let root = root.to_owned();
let deployment_id = deployment_id.to_owned();
blocking("deployment fetch", move || {
let workspace = ConfinedDir::open(&root)?;
deployment_sync(&workspace, &deployment_id)
})
.await
}
pub async fn bind_run(
root: &Path,
deployment_id: &str,
workflow_id: String,
run_id: String,
) -> Result<DeploymentRecord, RevisionError> {
validate_identifier(deployment_id, "deployment id")?;
let root = root.to_owned();
let deployment_id = deployment_id.to_owned();
blocking("deployment binding", move || {
let workspace = ConfinedDir::open(&root)?;
let mut record = deployment_sync(&workspace, &deployment_id)?;
record.workflow_id = Some(workflow_id);
record.run_id = Some(run_id);
let bytes = serde_json::to_vec_pretty(&record)
.map_err(|error| RevisionError::InvalidRecord(error.to_string()))?;
workspace.atomic_write(&deployment_path(&deployment_id), &bytes)?;
Ok(record)
})
.await
}
pub async fn current_drifted(
root: &Path,
record: &DeploymentRecord,
) -> Result<bool, RevisionError> {
let current = super::documents::read(root, &record.document_path)
.await
.map_err(|error| RevisionError::InvalidRecord(error.to_string()))?;
Ok(content_hash(¤t.source) != record.content_hash)
}
fn store_sync(workspace: &ConfinedDir, source: &str) -> Result<Revision, RevisionError> {
let revision = Revision {
content_hash: content_hash(source),
source: source.to_owned(),
};
let path = revision_path(&revision.content_hash);
match workspace.create_new(&path, source.as_bytes()) {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
let existing = workspace.read_to_string(&path)?;
if existing != source {
return Err(RevisionError::InvalidRecord(format!(
"hash collision at {}",
revision.content_hash
)));
}
}
Err(error) => return Err(RevisionError::Io(error)),
}
Ok(revision)
}
fn fetch_sync(workspace: &ConfinedDir, hash: &str) -> Result<Revision, RevisionError> {
validate_hash(hash)?;
let source = workspace
.read_to_string(&revision_path(hash))
.map_err(|error| {
if error.kind() == io::ErrorKind::NotFound {
RevisionError::NotFound(hash.to_owned())
} else {
RevisionError::Io(error)
}
})?;
if content_hash(&source) != hash {
return Err(RevisionError::InvalidRecord(format!(
"stored revision {hash} failed content verification"
)));
}
Ok(Revision {
content_hash: hash.to_owned(),
source,
})
}
fn deployment_sync(
workspace: &ConfinedDir,
deployment_id: &str,
) -> Result<DeploymentRecord, RevisionError> {
let bytes = workspace
.read(&deployment_path(deployment_id))
.map_err(|error| {
if error.kind() == io::ErrorKind::NotFound {
RevisionError::NotFound(deployment_id.to_owned())
} else {
RevisionError::Io(error)
}
})?;
let record: DeploymentRecord = serde_json::from_slice(&bytes)
.map_err(|error| RevisionError::InvalidRecord(error.to_string()))?;
validate_hash(&record.content_hash)?;
Ok(record)
}
fn revision_path(hash: &str) -> PathBuf {
Path::new(STATE_DIR).join(REVISION_DIR).join(hash)
}
fn deployment_path(deployment_id: &str) -> PathBuf {
Path::new(STATE_DIR)
.join(DEPLOYMENT_DIR)
.join(format!("{deployment_id}.json"))
}
async fn blocking<T: Send + 'static>(
operation: &'static str,
work: impl FnOnce() -> Result<T, RevisionError> + Send + 'static,
) -> Result<T, RevisionError> {
tokio::task::spawn_blocking(work)
.await
.map_err(|error| io::Error::other(format!("{operation} task failed: {error}")))?
}
fn validate_hash(hash: &str) -> Result<(), RevisionError> {
if hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit()) {
Ok(())
} else {
Err(RevisionError::InvalidHash(hash.to_owned()))
}
}
fn validate_identifier(value: &str, label: &str) -> Result<(), RevisionError> {
if !value.is_empty()
&& value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
{
Ok(())
} else {
Err(RevisionError::InvalidRecord(format!("invalid {label}")))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn revisions_are_content_addressed_immutable_and_private()
-> Result<(), Box<dyn std::error::Error>> {
let workspace = crate::test_support::private_tempdir()?;
let first = store(workspace.path(), "workflow first\n").await?;
let same = store(workspace.path(), "workflow first\n").await?;
let changed = store(workspace.path(), "workflow second\n").await?;
assert_eq!(first.content_hash, same.content_hash);
assert_ne!(first.content_hash, changed.content_hash);
assert_eq!(
fetch(workspace.path(), &first.content_hash).await?.source,
first.source
);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let state = workspace.path().join(STATE_DIR);
assert_eq!(
std::fs::metadata(&state)?.permissions().mode() & 0o777,
0o700
);
let file = state.join(REVISION_DIR).join(&first.content_hash);
assert_eq!(std::fs::metadata(file)?.permissions().mode() & 0o777, 0o600);
}
Ok(())
}
#[tokio::test]
async fn deployment_round_trip_and_drift_detection() -> Result<(), Box<dyn std::error::Error>> {
let workspace = crate::test_support::private_tempdir()?;
super::super::documents::write(
workspace.path(),
"flow.awl",
super::super::documents::PutDocumentRequest {
source: "workflow first\n".to_owned(),
},
)
.await?;
let revision = store(workspace.path(), "workflow first\n").await?;
let record = DeploymentRecord {
deployment_id: "deploy-1".to_owned(),
document_path: "flow.awl".to_owned(),
content_hash: revision.content_hash,
package_id: "package-1".to_owned(),
workflow_type: "flow".to_owned(),
task_queue: "worker".to_owned(),
workflow_id: None,
run_id: None,
};
record_deployment(workspace.path(), &record).await?;
assert_eq!(deployment(workspace.path(), "deploy-1").await?, record);
assert!(!current_drifted(workspace.path(), &record).await?);
Ok(())
}
#[cfg(unix)]
#[tokio::test]
async fn authoring_state_link_is_refused_without_outside_write()
-> Result<(), Box<dyn std::error::Error>> {
use std::os::unix::fs::symlink;
let sandbox = crate::test_support::private_tempdir()?;
let workspace = sandbox.path().join("workspace");
let outside = sandbox.path().join("outside");
std::fs::create_dir(&workspace)?;
std::fs::create_dir(&outside)?;
symlink(&outside, workspace.join(STATE_DIR))?;
assert!(store(&workspace, "workflow escaped\n").await.is_err());
assert!(std::fs::read_dir(&outside)?.next().is_none());
Ok(())
}
}