use std::env::{self, VarError};
use std::fs;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::path::{Component, Path, PathBuf};
use std::sync::mpsc::RecvTimeoutError;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use atelier_sdk_remote::RemoteFolder;
use sha2::{Digest, Sha256};
use atelier_sdk_diff::{
Address, Delta, DeltaKind, Diff, Fidelity, FormatPackage, PackageId, as_text, detect_package,
diff_lines,
};
use atelier_sdk_docx::DocxPackage;
use notify::{Event, RecursiveMode, Watcher};
use crate::config::{
Actor, InstructionFidelity, ROOT_MOUNT, Source, SourceKind, SyncPolicy, WorkspaceConfig,
read_workspace_config, resolve_actor, write_workspace_config,
};
use crate::coordination::{Coordination, LeaseClaim, RequestRow, SessionRow};
use crate::engine::{
DiffSides, Engine, FileBlob, LADDER_FILE_SIZE_MAX, LandOutcome, Side, StepBack,
};
use crate::error::{Error, config_err, engine_err};
use crate::journal::{Act, Journal, JournalEntry};
use crate::landing::{
Approval, GateOutcome, Landing, LandingRequest, RequestId, RequestState, Restore,
};
use crate::projection::ProjectionCache;
use crate::read::{ReadResult, window_size, window_text};
use crate::session::{Instruction, Session, SessionId, SessionState, SourceChange};
use crate::watch::{
STOP_TICK, WatchEvent, WatchStop, event_is_content, settle, watcher_failed, watcher_gone,
};
pub use crate::engine::Snapshot;
const CONTROL_DIR: &str = ".atelier";
const JOURNAL_FILE: &str = "journal.sqlite3";
const SESSIONS_DIR: &str = "sessions";
pub(crate) const SKIP_NAMES: [&str; 3] = [".atelier", ".jj", ".git"];
const LANDING_LEASE_POINT: &str = "landing";
const LANDED_BOOKMARK: &str = "atelier";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SyncOutcome {
Synced {
snapshot: String,
},
Parked {
snapshot: String,
},
}
fn unreachable_remote<T>() -> Result<T, Error> {
Err(Error::Engine(
"sync_source lost its remote handle; this is a bug".to_owned(),
))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PullOutcome {
Pulled {
snapshot: String,
},
Current,
}
enum SyncTarget {
Folder(PathBuf),
Remote(String),
}
const LANDING_LEASE_TTL_MS: i64 = 30_000;
pub struct Workspace {
root: PathBuf,
actor: Actor,
engine: Engine,
mounts: Vec<MountedSource>,
journal: Journal,
coordination: Coordination,
packages: Vec<Box<dyn FormatPackage>>,
projections: ProjectionCache,
}
struct MountedSource {
name: String,
engine: Engine,
branch: Option<String>,
}
impl Workspace {
pub fn init(path: impl AsRef<Path>) -> Result<Self, Error> {
let root = path.as_ref().to_path_buf();
let actor = resolve_actor()?;
let control = root.join(CONTROL_DIR);
if control.exists() {
return Err(Error::WorkspaceExists(root));
}
if let Some(ancestor) = enclosing_workspace(&root) {
return Err(Error::NestedWorkspace(ancestor));
}
fs::create_dir_all(&control)?;
let engine = Engine::init(&root, &actor, &[])?;
let config = WorkspaceConfig::new(workspace_name(&root));
write_workspace_config(&control, &config)?;
let journal = Journal::open(&control.join(JOURNAL_FILE))?;
let coordination = Coordination::open(&control.join(JOURNAL_FILE))?;
let mounts = Vec::new();
let workspace = Self {
root,
actor,
engine,
mounts,
journal,
coordination,
packages: builtin_packages(),
projections: ProjectionCache::new(&control),
};
let entry = workspace.entry(Act::WorkspaceInit, None)?;
workspace.journal.append(&entry)?;
Ok(workspace)
}
pub fn open(path: impl AsRef<Path>) -> Result<Self, Error> {
let root = path.as_ref().to_path_buf();
let actor = resolve_actor()?;
let control = root.join(CONTROL_DIR);
if !control.exists() {
return Err(Error::NotAWorkspace(root));
}
let config = read_workspace_config(&control)?;
let mount_names = mount_names(&config);
let engine = Engine::open(&root, &actor, &mount_names)?;
let mut mounts = Vec::new();
for name in &mount_names {
let branch = config
.sources
.iter()
.find(|source| source.mount == *name)
.and_then(|source| source.branch.clone());
mounts.push(MountedSource {
name: name.clone(),
engine: Engine::open(&root.join(name), &actor, &[])?,
branch,
});
}
let journal = Journal::open(&control.join(JOURNAL_FILE))?;
let coordination = Coordination::open(&control.join(JOURNAL_FILE))?;
Ok(Self {
root,
actor,
engine,
mounts,
journal,
coordination,
packages: builtin_packages(),
projections: ProjectionCache::new(&control),
})
}
#[must_use]
pub fn actor(&self) -> &Actor {
&self.actor
}
pub fn attach(&mut self, folder: impl AsRef<Path>) -> Result<Source, Error> {
let folder = folder.as_ref();
if !folder.is_dir() {
return Err(Error::Io(std::io::Error::new(
std::io::ErrorKind::NotFound,
format!("source folder not found: {}", folder.display()),
)));
}
let control = self.root.join(CONTROL_DIR);
let mut config = read_workspace_config(&control)?;
if config.sources.iter().any(|s| s.mount == ROOT_MOUNT) {
return Err(Error::AlreadyAttached);
}
if folder_uses_lfs(folder)? {
return Err(Error::LfsSourceUnsupported);
}
self.auto_snapshot()?;
copy_tree(folder, &self.root, &SKIP_NAMES)?;
let source = Source {
kind: SourceKind::LocalFolder,
path: folder.to_path_buf(),
sync: SyncPolicy::TwoWay,
mount: ROOT_MOUNT.to_owned(),
branch: None,
};
config.sources.push(source.clone());
write_workspace_config(&control, &config)?;
let snapshot = self.engine.snapshot()?;
let fingerprint = folder_fingerprint(folder)?;
self.coordination
.record_sync_state(ROOT_MOUNT, &fingerprint, &self.engine.head()?)?;
let entry = self.entry(Act::SourceAttach, snapshot)?;
self.journal.append(&entry)?;
Ok(source)
}
pub fn attach_mount(&mut self, folder: impl AsRef<Path>, name: &str) -> Result<Source, Error> {
let folder = folder.as_ref();
if !folder.is_dir() {
return Err(Error::Io(std::io::Error::new(
std::io::ErrorKind::NotFound,
format!("source folder not found: {}", folder.display()),
)));
}
let control = self.root.join(CONTROL_DIR);
let mut config = read_workspace_config(&control)?;
let mount_dir = self.root.join(name);
mount_refusals(name, &config, &mount_dir)?;
if folder_uses_lfs(folder)? {
return Err(Error::LfsSourceUnsupported);
}
self.auto_snapshot()?;
let (kind, engine, snapshot) = match self.import_folder(folder, &mount_dir) {
Ok(imported) => imported,
Err(error) => {
let _ = fs::remove_dir_all(&mount_dir);
return Err(error);
}
};
let branch = match kind {
SourceKind::LocalGit => adopted_branch(folder)?,
SourceKind::LocalFolder | SourceKind::Remote => None,
};
let source = Source {
kind,
path: folder.to_path_buf(),
sync: SyncPolicy::TwoWay,
mount: name.to_owned(),
branch: branch.clone(),
};
config.sources.push(source.clone());
write_workspace_config(&control, &config)?;
let mount_names = mount_names(&config);
self.engine = Engine::open(&self.root, &self.actor, &mount_names)?;
let position = self
.mounts
.binary_search_by(|mount| mount.name.as_str().cmp(name))
.unwrap_or_else(|position| position);
self.mounts.insert(
position,
MountedSource {
name: name.to_owned(),
engine,
branch,
},
);
match kind {
SourceKind::LocalFolder => {
let fingerprint = folder_fingerprint(folder)?;
let head = self.mounts[position].engine.head()?;
self.coordination
.record_sync_state(name, &fingerprint, &head)?;
}
SourceKind::LocalGit | SourceKind::Remote => {}
}
let reference = snapshot.map(|id| format!("{name} {id}"));
let entry = self.entry(Act::SourceAttach, reference)?;
self.journal.append(&entry)?;
Ok(source)
}
fn import_folder(
&self,
folder: &Path,
mount_dir: &Path,
) -> Result<(SourceKind, Engine, Option<String>), Error> {
fs::create_dir_all(mount_dir)?;
let adopts_git = folder.join(".git").is_dir();
let (kind, mut engine) = if adopts_git {
copy_tree(folder, mount_dir, &[".atelier", ".jj"])?;
(
SourceKind::LocalGit,
Engine::adopt_git(mount_dir, &self.actor, &[])?,
)
} else {
let engine = Engine::init(mount_dir, &self.actor, &[])?;
copy_tree(folder, mount_dir, &SKIP_NAMES)?;
(SourceKind::LocalFolder, engine)
};
let snapshot = engine.snapshot()?;
Ok((kind, engine, snapshot))
}
fn import_remote(
&self,
remote: &RemoteFolder,
mount_dir: &Path,
) -> Result<(Engine, Option<String>), Error> {
fs::create_dir_all(mount_dir)?;
let mut engine = Engine::init(mount_dir, &self.actor, &[])?;
remote.download_all(mount_dir).map_err(engine_err)?;
let snapshot = engine.snapshot()?;
Ok((engine, snapshot))
}
pub fn attach_remote(&mut self, url: &str, name: &str) -> Result<Source, Error> {
let control = self.root.join(CONTROL_DIR);
let mut config = read_workspace_config(&control)?;
let mount_dir = self.root.join(name);
mount_refusals(name, &config, &mount_dir)?;
self.auto_snapshot()?;
let remote = RemoteFolder::open(url).map_err(engine_err)?;
let (engine, snapshot) = match self.import_remote(&remote, &mount_dir) {
Ok(imported) => imported,
Err(error) => {
let _ = fs::remove_dir_all(&mount_dir);
return Err(error);
}
};
let fingerprint = remote.fingerprint().map_err(engine_err)?;
let head = engine.head()?;
self.coordination
.record_sync_state(name, &fingerprint, &head)?;
let source = Source {
kind: SourceKind::Remote,
path: PathBuf::from(url),
sync: SyncPolicy::TwoWay,
mount: name.to_owned(),
branch: None,
};
config.sources.push(source.clone());
write_workspace_config(&control, &config)?;
let mount_names = mount_names(&config);
self.engine = Engine::open(&self.root, &self.actor, &mount_names)?;
let position = self
.mounts
.binary_search_by(|mount| mount.name.as_str().cmp(name))
.unwrap_or_else(|position| position);
self.mounts.insert(
position,
MountedSource {
name: name.to_owned(),
engine,
branch: None,
},
);
let reference = snapshot.map(|id| format!("{name} {id}"));
let entry = self.entry(Act::SourceAttach, reference)?;
self.journal.append(&entry)?;
Ok(source)
}
pub fn log(&mut self, limit: usize) -> Result<Vec<SourceSnapshot>, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let mut entries = Vec::new();
for snapshot in self.engine.log(limit)? {
entries.push(SourceSnapshot {
source: None,
snapshot,
});
}
for mount in &self.mounts {
for snapshot in mount.engine.log(limit)? {
entries.push(SourceSnapshot {
source: Some(mount.name.clone()),
snapshot,
});
}
}
Ok(entries)
}
pub fn diff_latest(&mut self) -> Result<Diff, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let (diff, sides) = self.engine.diff_latest()?;
let mut deltas = self.raised(&self.engine, diff, &sides, None)?.deltas;
for mount in &self.mounts {
let (mount_diff, sides) = mount.engine.diff_latest()?;
let raised = self.raised(&mount.engine, mount_diff, &sides, Some(&mount.name))?;
deltas.extend(raised.deltas);
}
Ok(Diff { deltas })
}
pub fn manifest(&mut self) -> Result<String, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let config = read_workspace_config(&self.root.join(CONTROL_DIR))?;
let mut lines = vec![
format!("workspace: {}", config.workspace.name),
format!("schema: {}", config.schema),
String::new(),
"sources:".to_owned(),
];
if config.sources.is_empty() {
lines.push(" (none)".to_owned());
}
for source in &config.sources {
lines.push(format!(
" {} {} {} {}",
source.mount,
source.kind,
source.path.display(),
source.sync
));
}
lines.push(String::new());
lines.push("discipline:".to_owned());
let landing = config.landing;
let self_approve = if landing.allow_self_approve {
"allowed"
} else {
"forbidden"
};
let dismiss = if landing.dismiss_approvals_on_new_snapshots {
"yes"
} else {
"no"
};
lines.push(format!(
" approvals: {} self-approval: {self_approve} snapshots dismiss approvals: {dismiss}",
landing.approvals
));
let fidelity = match config.journal.instruction_fidelity {
InstructionFidelity::Summary => "summary",
InstructionFidelity::Verbatim => "verbatim",
};
lines.push(format!(" instructions: {fidelity}"));
lines.push(String::new());
lines.push("state:".to_owned());
for line in self.state_lines()? {
lines.push(format!(" {line}"));
}
lines.push(String::new());
lines.push("the loop:".to_owned());
lines
.push(" open_session -> write -> diff -> land (or request_land + approve)".to_owned());
lines.push(
" mount-scoped paths address sources; editing never takes the landing lease"
.to_owned(),
);
Ok(lines.join("\n"))
}
pub fn status(&mut self) -> Result<String, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
Ok(self.state_lines()?.join("\n"))
}
fn state_lines(&mut self) -> Result<Vec<String>, Error> {
let mut lines = vec![format!("head: {}", self.engine.head()?)];
for mount in &self.mounts {
lines.push(format!("head {}: {}", mount.name, mount.engine.head()?));
}
let mut open_sessions: Vec<String> = self
.sessions()?
.into_iter()
.filter(|session| match session.state {
SessionState::Open => true,
SessionState::Landed | SessionState::Abandoned => false,
})
.map(|session| session.id.to_string())
.collect();
open_sessions.reverse();
lines.push(if open_sessions.is_empty() {
"open sessions: none".to_owned()
} else {
format!("open sessions: {}", open_sessions.join(", "))
});
let mut live_requests: Vec<String> = self
.landing_requests()?
.into_iter()
.filter(|request| match request.state {
RequestState::Open | RequestState::Approved | RequestState::Parked => true,
RequestState::Landed | RequestState::Rejected | RequestState::Abandoned => false,
})
.map(|request| format!("{} ({})", request.id, request.state))
.collect();
live_requests.reverse();
lines.push(if live_requests.is_empty() {
"live requests: none".to_owned()
} else {
format!("live requests: {}", live_requests.join(", "))
});
Ok(lines)
}
pub fn diff_between(&mut self, before: &str, after: &str) -> Result<Diff, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let (diff, sides) = self.engine.diff_between(before, after)?;
self.raised(&self.engine, diff, &sides, None)
}
pub fn journal(&mut self, limit: usize) -> Result<Vec<JournalEntry>, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
self.journal.entries(limit)
}
pub fn open_session(
&mut self,
actor: &Actor,
instruction: &Instruction,
) -> Result<Session, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let verbatim = match self.config()?.journal.instruction_fidelity {
InstructionFidelity::Summary => None,
InstructionFidelity::Verbatim => instruction.verbatim.clone(),
};
let row = self.coordination.create_session(
actor,
&instruction.summary,
instruction.run_ref.as_deref(),
verbatim.as_deref(),
now_ms()?,
)?;
let id = SessionId(row);
let change_id = match self.engine.create_session_workspace(
&self.session_root(id),
&format!("session-{id}"),
actor,
) {
Ok(change_id) => change_id,
Err(error) => {
self.coordination.delete_session(row)?;
return Err(error);
}
};
self.coordination.set_session_change(row, &change_id)?;
let session_root = self.session_root(id);
for index in 0..self.mounts.len() {
let name = self.mounts[index].name.clone();
let mount_change = match self.mounts[index].engine.create_session_workspace(
&session_root.join(&name),
&format!("session-{id}"),
actor,
) {
Ok(change_id) => change_id,
Err(error) => {
self.coordination.delete_session(row)?;
return Err(error);
}
};
self.coordination
.set_session_source_change(row, &name, &mount_change)?;
}
self.journal.append(&JournalEntry {
at_ms: now_ms()?,
actor_name: actor.name.clone(),
actor_kind: actor.kind,
act: Act::SessionOpen,
session: Some(id.to_string()),
instruction_summary: Some(instruction.summary.clone()),
instruction_run_ref: instruction.run_ref.clone(),
instruction_verbatim: verbatim,
reference: None,
})?;
self.session(id)
}
pub fn sessions(&mut self) -> Result<Vec<Session>, Error> {
let rows = self.coordination.sessions()?;
rows.into_iter().map(|row| self.session_from(row)).collect()
}
pub fn session(&mut self, id: SessionId) -> Result<Session, Error> {
match self.coordination.session(id.0)? {
Some(row) => self.session_from(row),
None => Err(Error::SessionNotFound(id.to_string())),
}
}
pub fn session_write(
&mut self,
id: SessionId,
path: &str,
content: &str,
) -> Result<String, Error> {
self.engine.refresh()?;
let session = self.open_session_only(id)?;
let (source, directory, inner) = self.session_target(&session, path);
let file = session_file(&directory, &inner)?;
if let Some(parent) = file.parent() {
fs::create_dir_all(parent)?;
}
fs::write(&file, content)?;
let tips = self.snapshot_session(&session)?;
Ok(tips.tip_of(source.as_deref()))
}
pub fn session_read(
&mut self,
id: SessionId,
path: &str,
start: usize,
max_bytes: Option<usize>,
) -> Result<ReadResult, Error> {
let session = self.open_session_only(id)?;
let size = window_size(max_bytes)?;
let (_, directory, inner) = self.session_target(&session, path);
let file = session_file(&directory, &inner)?;
let bytes = fs::read(&file)?;
if let Some(package) = self.detected(path, &bytes)? {
let text = self.project_for_read(package, &bytes)?;
return Ok(window_text(&text, start, size, Some(package.id())));
}
match as_text(&bytes) {
Some(text) => Ok(window_text(text, start, size, None)),
None => Err(Error::NotText(path.to_owned())),
}
}
pub fn session_diff(&mut self, id: SessionId) -> Result<Diff, Error> {
self.refresh_engines()?;
let session = self.open_session_only(id)?;
let tips = self.snapshot_session(&session)?;
let base = self.engine.parent_of(&tips.root)?;
let (diff, sides) = self.engine.diff_between(&base, &tips.root)?;
let mut deltas = self.raised(&self.engine, diff, &sides, None)?.deltas;
for (name, tip) in &tips.mounts {
let mount = self.mount(name)?;
let base = mount.engine.parent_of(tip)?;
let (diff, sides) = mount.engine.diff_between(&base, tip)?;
let raised = self.raised(&mount.engine, diff, &sides, Some(name))?;
deltas.extend(raised.deltas);
}
Ok(Diff { deltas })
}
pub fn request_land(&mut self, id: SessionId) -> Result<LandingRequest, Error> {
self.refresh_engines()?;
let session = self.open_session_only(id)?;
self.snapshot_session(&session)?;
if let Some(row) = self.coordination.gated_request_for_session(id.0)? {
return self.request_from(row);
}
let row = self
.coordination
.create_request(id.0, &session.actor, now_ms()?)?;
let request_id = RequestId(row);
self.append_session_entry(
&session.actor,
Act::LandRequest,
id,
Some(request_id.to_string()),
)?;
self.request(request_id)
}
pub fn landing_requests(&mut self) -> Result<Vec<LandingRequest>, Error> {
let rows = self.coordination.requests()?;
rows.into_iter().map(|row| self.request_from(row)).collect()
}
pub fn request(&mut self, id: RequestId) -> Result<LandingRequest, Error> {
match self.coordination.request(id.0)? {
Some(row) => self.request_from(row),
None => Err(Error::RequestNotFound(id.to_string())),
}
}
pub fn approve(&mut self, id: RequestId, approver: &Actor) -> Result<GateOutcome, Error> {
self.refresh_engines()?;
let row = self.gated_request(id)?;
let session = self.open_session_only(SessionId(row.session_id))?;
let policy = self.config()?.landing;
let requester = Actor {
name: row.requester_name.clone(),
kind: row.requester_kind,
};
if !policy.allow_self_approve && *approver == requester {
return Err(Error::SelfApprovalForbidden);
}
let tips = self.snapshot_session(&session)?;
let tip = tips.root.clone();
let row = self.gated_request(id)?;
if let RequestState::Open = row.state {
self.coordination
.add_approval(row.id, approver, &tip, now_ms()?)?;
self.append_session_entry(
approver,
Act::Approve,
session.id,
Some(format!("{id} {tip}")),
)?;
}
let approvals = self.coordination.live_approvals(row.id)?;
let approvers: std::collections::BTreeSet<(&str, &str)> = approvals
.iter()
.map(|approval| (approval.actor_name.as_str(), approval.actor_kind.as_str()))
.collect();
if (approvers.len() as u64) < u64::from(policy.approvals) {
return Ok(GateOutcome::Pending {
request: self.request(id)?,
required: policy.approvals,
});
}
if !self.coordination.move_request_state(
row.id,
&[RequestState::Open],
RequestState::Approved,
)? {
let row = self.gated_request(id)?;
if let RequestState::Open = row.state {
return Ok(GateOutcome::Pending {
request: self.request(id)?,
required: policy.approvals,
});
}
}
self.apply(&session, id, &tips, approver)
}
pub fn reject(
&mut self,
id: RequestId,
actor: &Actor,
reason: Option<&str>,
) -> Result<LandingRequest, Error> {
let row = self.gated_request(id)?;
while !self.coordination.move_request_state(
row.id,
&[RequestState::Open, RequestState::Approved],
RequestState::Rejected,
)? {
self.gated_request(id)?;
}
let reference = match reason {
Some(reason) => format!("{id} {reason}"),
None => id.to_string(),
};
self.append_session_entry(
actor,
Act::Reject,
SessionId(row.session_id),
Some(reference),
)?;
self.request(id)
}
pub fn land(&mut self, id: SessionId) -> Result<GateOutcome, Error> {
let request = self.request_land(id)?;
let session = self.session(id)?;
let policy = self.config()?.landing;
if !policy.allow_self_approve {
return Ok(GateOutcome::Pending {
request,
required: policy.approvals,
});
}
self.approve(request.id, &session.actor)
}
pub fn abandon(&mut self, id: SessionId) -> Result<Session, Error> {
self.engine.refresh()?;
let session = self.open_session_only(id)?;
self.snapshot_session(&session)?;
let mut reference = None;
if let Some(request) = self.coordination.gated_request_for_session(id.0)? {
let _ = self.coordination.move_request_state(
request.id,
&[
RequestState::Open,
RequestState::Approved,
RequestState::Parked,
],
RequestState::Abandoned,
)?;
reference = Some(RequestId(request.id).to_string());
}
if !self.coordination.move_session_state(
id.0,
SessionState::Open,
SessionState::Abandoned,
)? {
let session = self.session(id)?;
return Err(Error::SessionClosed {
id: id.to_string(),
state: session.state.to_string(),
});
}
self.append_session_entry(&session.actor, Act::SessionAbandon, id, reference)?;
self.session(id)
}
fn auto_snapshot(&mut self) -> Result<Vec<(Option<String>, String)>, Error> {
let mut recorded = Vec::new();
if let Some(id) = self.engine.snapshot()? {
let entry = self.entry(Act::Snapshot, Some(id.clone()))?;
self.journal.append(&entry)?;
recorded.push((None, id));
}
for mount in &mut self.mounts {
if let Some(id) = mount.engine.snapshot()? {
recorded.push((Some(mount.name.clone()), id));
}
}
for (mount, id) in &recorded {
if let Some(mount) = mount {
let entry = self.entry(Act::Snapshot, Some(format!("{mount} {id}")))?;
self.journal.append(&entry)?;
}
}
Ok(recorded)
}
fn refresh_engines(&mut self) -> Result<(), Error> {
self.engine.refresh()?;
for mount in &mut self.mounts {
mount.engine.refresh()?;
}
Ok(())
}
pub fn watch(
&mut self,
debounce: Duration,
mut on_event: impl FnMut(&WatchEvent),
stop: &WatchStop,
) -> Result<(), Error> {
let root = fs::canonicalize(&self.root)?;
let (pulses, storm) = std::sync::mpsc::channel();
let filter_root = root.clone();
let mut watcher =
notify::recommended_watcher(move |event: Result<Event, notify::Error>| {
let pulse = match event {
Ok(event) => {
if !event_is_content(&filter_root, &event) {
return;
}
Ok(())
}
Err(error) => Err(error),
};
let _ = pulses.send(pulse);
})
.map_err(|error| watcher_failed(&error))?;
watcher
.watch(&root, RecursiveMode::Recursive)
.map_err(|error| watcher_failed(&error))?;
on_event(&WatchEvent::Started);
self.snapshot_watched(&mut on_event)?;
while !stop.stopped() {
match storm.recv_timeout(STOP_TICK) {
Ok(Ok(())) => {
settle(&storm, debounce, stop)?;
self.snapshot_watched(&mut on_event)?;
}
Ok(Err(error)) => return Err(watcher_failed(&error)),
Err(RecvTimeoutError::Timeout) => {}
Err(RecvTimeoutError::Disconnected) => return Err(watcher_gone()),
}
}
Ok(())
}
fn snapshot_watched(&mut self, on_event: &mut impl FnMut(&WatchEvent)) -> Result<(), Error> {
self.refresh_engines()?;
for (_, snapshot) in self.auto_snapshot()? {
on_event(&WatchEvent::Snapshotted { snapshot });
}
Ok(())
}
fn apply(
&mut self,
session: &Session,
id: RequestId,
tips: &SessionTips,
approver: &Actor,
) -> Result<GateOutcome, Error> {
let already = self.coordination.landings(id.0)?;
let mut parked = Vec::new();
let mut plan: Vec<(Option<String>, String)> = vec![(None, tips.root.clone())];
for (name, tip) in &tips.mounts {
if self.mount(name)?.engine.tree_changed(tip)? {
plan.push((Some(name.clone()), tip.clone()));
}
}
for (source, tip) in plan {
if already.iter().any(|(landed, _)| *landed == source) {
continue;
}
match self.apply_source(session, id, source.as_deref(), &tip, approver)? {
LandOutcome::Landed { .. } => {}
LandOutcome::Conflicted => parked.push(source),
}
}
let landings: Vec<Landing> = self
.coordination
.landings(id.0)?
.into_iter()
.map(|(source, snapshot)| Landing { source, snapshot })
.collect();
if parked.is_empty() {
if self.coordination.move_request_state(
id.0,
&[RequestState::Approved],
RequestState::Landed,
)? {
let _ = self.coordination.move_session_state(
session.id.0,
SessionState::Open,
SessionState::Landed,
)?;
}
return Ok(GateOutcome::Landed { landings });
}
let _ = self.coordination.move_request_state(
id.0,
&[RequestState::Approved],
RequestState::Parked,
)?;
Ok(GateOutcome::Parked {
request: self.request(id)?,
landings,
parked,
})
}
fn apply_source(
&mut self,
session: &Session,
id: RequestId,
source: Option<&str>,
tip: &str,
approver: &Actor,
) -> Result<LandOutcome, Error> {
let point = match source {
Some(name) => format!("{LANDING_LEASE_POINT}/{name}"),
None => LANDING_LEASE_POINT.to_owned(),
};
let holder = format!("{}:{}", self.actor.name, std::process::id());
let now = now_ms()?;
match self
.coordination
.claim_lease(&point, &holder, now, LANDING_LEASE_TTL_MS)?
{
LeaseClaim::HeldByOther {
holder,
expires_at_ms,
} => {
return Err(Error::LeaseHeld {
holder,
expires_at_ms,
});
}
LeaseClaim::Held => {}
}
let outcome = self.apply_source_holding_lease(session, id, source, tip, approver);
let released = self.coordination.release_lease(&point, &holder);
let outcome = outcome?;
released?;
Ok(outcome)
}
fn apply_source_holding_lease(
&mut self,
session: &Session,
id: RequestId,
source: Option<&str>,
tip: &str,
approver: &Actor,
) -> Result<LandOutcome, Error> {
if let Some(hold) = land_hold_ms()? {
std::thread::sleep(Duration::from_millis(hold));
}
self.refresh_engines()?;
self.auto_snapshot()?;
let outcome = match source {
None => self.engine.land(tip, LANDED_BOOKMARK)?,
Some(name) => {
let index = self
.mounts
.iter()
.position(|mount| mount.name == name)
.ok_or_else(|| Error::Engine(format!("no source is mounted at {name:?}")))?;
let bookmark = self.mounts[index]
.branch
.clone()
.unwrap_or_else(|| LANDED_BOOKMARK.to_owned());
self.mounts[index].engine.land(tip, &bookmark)?
}
};
let scoped = |text: &str| match source {
Some(name) => format!("{name} {text}"),
None => text.to_owned(),
};
match &outcome {
LandOutcome::Conflicted => {
self.append_session_entry(
approver,
Act::LandParked,
session.id,
Some(scoped(&id.to_string())),
)?;
}
LandOutcome::Landed { snapshot } => {
self.coordination.record_landing(id.0, source, snapshot)?;
self.append_session_entry(
approver,
Act::Land,
session.id,
Some(scoped(&format!("{id} {snapshot}"))),
)?;
self.sync_after_line_move(session, source, approver)?;
}
}
Ok(outcome)
}
pub fn undo(&mut self, id: RequestId) -> Result<Vec<Restore>, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let Some(row) = self.coordination.request(id.0)? else {
return Err(Error::RequestNotFound(id.to_string()));
};
match row.state {
RequestState::Landed => {}
RequestState::Open
| RequestState::Approved
| RequestState::Parked
| RequestState::Rejected
| RequestState::Abandoned => {
return Err(Error::Config(format!(
"{id} is {}; only a landed request undoes - snapshots amend forward, gate acts move forward, syncs reconcile with atelier sync",
row.state
)));
}
}
let session = self.session(SessionId(row.session_id))?;
let mut landings = self.coordination.landings(id.0)?;
landings.reverse();
let mut restores = Vec::new();
for (source, landed) in &landings {
if let Some(head) = self.undo_source(&session, id, source.as_deref(), landed)? {
restores.push(Restore {
source: source.clone(),
head,
});
}
}
if self.coordination.move_request_state(
id.0,
&[RequestState::Landed],
RequestState::Open,
)? {
let actor = self.actor.clone();
if self.coordination.dismiss_approvals(id.0)? > 0 {
self.append_session_entry(
&actor,
Act::ApprovalsDismissed,
session.id,
Some(id.to_string()),
)?;
}
let _ = self.coordination.move_session_state(
session.id.0,
SessionState::Landed,
SessionState::Open,
)?;
}
Ok(restores)
}
fn undo_source(
&mut self,
session: &Session,
id: RequestId,
source: Option<&str>,
landed: &str,
) -> Result<Option<String>, Error> {
let point = match source {
Some(name) => format!("{LANDING_LEASE_POINT}/{name}"),
None => LANDING_LEASE_POINT.to_owned(),
};
let holder = format!("{}:{}", self.actor.name, std::process::id());
let now = now_ms()?;
match self
.coordination
.claim_lease(&point, &holder, now, LANDING_LEASE_TTL_MS)?
{
LeaseClaim::HeldByOther {
holder,
expires_at_ms,
} => {
return Err(Error::LeaseHeld {
holder,
expires_at_ms,
});
}
LeaseClaim::Held => {}
}
let outcome = self.undo_source_holding_lease(session, id, source, landed);
let released = self.coordination.release_lease(&point, &holder);
let outcome = outcome?;
released?;
Ok(outcome)
}
fn undo_source_holding_lease(
&mut self,
session: &Session,
id: RequestId,
source: Option<&str>,
landed: &str,
) -> Result<Option<String>, Error> {
let step = match source {
None => self.engine.step_back(landed, LANDED_BOOKMARK)?,
Some(name) => {
let index = self
.mounts
.iter()
.position(|mount| mount.name == name)
.ok_or_else(|| Error::Engine(format!("no source is mounted at {name:?}")))?;
let bookmark = self.mounts[index]
.branch
.clone()
.unwrap_or_else(|| LANDED_BOOKMARK.to_owned());
self.mounts[index].engine.step_back(landed, &bookmark)?
}
};
match step {
StepBack::Stepped { restored } => {
self.coordination.delete_landing(id.0, source)?;
let reference = match source {
Some(name) => format!("{name} {id} {restored}"),
None => format!("{id} {restored}"),
};
let actor = self.actor.clone();
self.append_session_entry(&actor, Act::Undo, session.id, Some(reference))?;
self.sync_after_line_move(session, source, &actor)?;
Ok(Some(restored))
}
StepBack::AlreadyStepped => {
self.coordination.delete_landing(id.0, source)?;
Ok(None)
}
StepBack::LineMoved { head } => {
let line = source.unwrap_or("the root");
Err(Error::Config(format!(
"{line} moved past {id}: {head} sits on the line now; undo that landing first"
)))
}
}
}
pub fn pull(&mut self, source: Option<&str>) -> Result<PullOutcome, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let mount = source.unwrap_or(ROOT_MOUNT);
let config = read_workspace_config(&self.root.join(CONTROL_DIR))?;
let Some(entry) = config.sources.iter().find(|s| s.mount == mount) else {
return Err(Error::Config(format!("no source is attached at {mount:?}")));
};
match entry.kind {
SourceKind::Remote => {}
SourceKind::LocalFolder | SourceKind::LocalGit => {
return Err(Error::Config(format!(
"{mount:?} is not a remote source; folders reconcile with atelier sync and git sources pull with plain git"
)));
}
}
let index = self
.mounts
.iter()
.position(|m| m.name == mount)
.ok_or_else(|| Error::Engine(format!("no source is mounted at {mount:?}")))?;
let remote = RemoteFolder::open(&entry.path.display().to_string()).map_err(engine_err)?;
let Some((recorded, last_synced)) = self.coordination.sync_state(mount)? else {
return Err(Error::Config(format!(
"{mount:?} has no sync record; atelier sync --force seeds one"
)));
};
if remote.fingerprint().map_err(engine_err)? == recorded {
return Ok(PullOutcome::Current);
}
let head = self.mounts[index].engine.head()?;
if head != last_synced {
return Err(Error::Config(format!(
"{mount:?} moved locally since its last sync ({head}); land or sync it first, then pull"
)));
}
let mount_dir = self.root.join(mount);
remote.download_mirror(&mount_dir).map_err(engine_err)?;
let Some(snapshot) = self.mounts[index].engine.snapshot()? else {
let fingerprint = remote.fingerprint().map_err(engine_err)?;
self.coordination
.record_sync_state(mount, &fingerprint, &head)?;
return Ok(PullOutcome::Current);
};
let fingerprint = remote.fingerprint().map_err(engine_err)?;
self.coordination
.record_sync_state(mount, &fingerprint, &snapshot)?;
let entry = self.entry(Act::Pull, Some(format!("{mount} {snapshot}")))?;
self.journal.append(&entry)?;
Ok(PullOutcome::Pulled { snapshot })
}
pub fn sync(&mut self, source: Option<&str>, force: bool) -> Result<SyncOutcome, Error> {
self.refresh_engines()?;
self.auto_snapshot()?;
let mount = source.unwrap_or(ROOT_MOUNT);
let config = read_workspace_config(&self.root.join(CONTROL_DIR))?;
let Some(entry) = config.sources.iter().find(|s| s.mount == mount) else {
return Err(Error::Config(format!("no source is attached at {mount:?}")));
};
let target = match entry.kind {
SourceKind::LocalGit => {
return Err(Error::Config(format!(
"{mount:?} is a git source; landed work publishes with plain git push"
)));
}
SourceKind::LocalFolder => SyncTarget::Folder(self.origin_path(&entry.path)),
SourceKind::Remote => SyncTarget::Remote(entry.path.display().to_string()),
};
let outcome = self.sync_source(source, &target, force)?;
let (act, detail) = match &outcome {
SyncOutcome::Synced { snapshot } => (Act::Sync, snapshot.clone()),
SyncOutcome::Parked { snapshot } => {
(Act::SyncParked, format!("{snapshot} origin changed"))
}
};
let reference = match source {
Some(name) => format!("{name} {detail}"),
None => detail,
};
let entry = self.entry(act, Some(reference))?;
self.journal.append(&entry)?;
Ok(outcome)
}
fn origin_path(&self, configured: &Path) -> PathBuf {
if configured.is_absolute() {
configured.to_path_buf()
} else {
self.root.join(configured)
}
}
fn sync_source(
&mut self,
source: Option<&str>,
target: &SyncTarget,
force: bool,
) -> Result<SyncOutcome, Error> {
let mount = source.unwrap_or(ROOT_MOUNT);
let index = match source {
None => None,
Some(name) => Some(
self.mounts
.iter()
.position(|m| m.name == name)
.ok_or_else(|| Error::Engine(format!("no source is mounted at {name:?}")))?,
),
};
let snapshot = match index {
None => self.engine.head()?,
Some(index) => self.mounts[index].engine.head()?,
};
let remote = match target {
SyncTarget::Folder(_) => None,
SyncTarget::Remote(url) => Some(RemoteFolder::open(url).map_err(engine_err)?),
};
if !force {
let recorded = self.coordination.sync_state(mount)?;
let current = match (target, &remote) {
(SyncTarget::Folder(origin), _) => folder_fingerprint(origin)?,
(SyncTarget::Remote(_), Some(remote)) => {
remote.fingerprint().map_err(engine_err)?
}
(SyncTarget::Remote(_), None) => unreachable_remote()?,
};
if recorded.map(|(fingerprint, _)| fingerprint) != Some(current) {
return Ok(SyncOutcome::Parked { snapshot });
}
}
let engine = match index {
None => &self.engine,
Some(index) => &self.mounts[index].engine,
};
let fingerprint = match (target, &remote) {
(SyncTarget::Folder(origin), _) => {
engine.export_tree(&snapshot, origin)?;
folder_fingerprint(origin)?
}
(SyncTarget::Remote(_), Some(remote)) => {
let scratch = tempfile::tempdir()?;
engine.export_tree(&snapshot, scratch.path())?;
remote.mirror(scratch.path()).map_err(engine_err)?;
remote.fingerprint().map_err(engine_err)?
}
(SyncTarget::Remote(_), None) => unreachable_remote()?,
};
self.coordination
.record_sync_state(mount, &fingerprint, &snapshot)?;
Ok(SyncOutcome::Synced { snapshot })
}
fn sync_after_line_move(
&mut self,
session: &Session,
source: Option<&str>,
approver: &Actor,
) -> Result<(), Error> {
let mount = source.unwrap_or(ROOT_MOUNT);
let config = read_workspace_config(&self.root.join(CONTROL_DIR))?;
let Some(entry) = config.sources.iter().find(|s| s.mount == mount) else {
return Ok(());
};
let target = match entry.kind {
SourceKind::LocalGit => return Ok(()),
SourceKind::LocalFolder => SyncTarget::Folder(self.origin_path(&entry.path)),
SourceKind::Remote => SyncTarget::Remote(entry.path.display().to_string()),
};
let (act, detail) = match self.sync_source(source, &target, false) {
Ok(SyncOutcome::Synced { snapshot }) => (Act::Sync, snapshot),
Ok(SyncOutcome::Parked { snapshot }) => {
(Act::SyncParked, format!("{snapshot} origin changed"))
}
Err(error) => (Act::SyncParked, error.to_string()),
};
let reference = match source {
Some(name) => format!("{name} {detail}"),
None => detail,
};
self.append_session_entry(approver, act, session.id, Some(reference))?;
Ok(())
}
fn snapshot_session(&mut self, session: &Session) -> Result<SessionTips, Error> {
let boundary = self.mount_boundary();
let mut engine = Engine::open(&session.working_copy, &session.actor, &boundary)?;
let new_snapshot = engine.snapshot_amend()?;
let root_tip = engine.head()?;
let mut recorded: Vec<(Option<String>, String)> = Vec::new();
if let Some(new_snapshot) = new_snapshot {
recorded.push((None, new_snapshot));
}
let mut mounts = Vec::new();
for name in boundary {
let mut engine = Engine::open(&session.working_copy.join(&name), &session.actor, &[])?;
if let Some(new_snapshot) = engine.snapshot_amend()? {
recorded.push((Some(name.clone()), new_snapshot));
}
mounts.push((name, engine.head()?));
}
for (source, new_snapshot) in &recorded {
let reference = match source {
Some(source) => format!("{source} {new_snapshot}"),
None => new_snapshot.clone(),
};
self.append_session_entry(&session.actor, Act::Snapshot, session.id, Some(reference))?;
self.gate_reacts_to_snapshot(session, new_snapshot)?;
}
if !recorded.is_empty() {
self.refresh_engines()?;
}
Ok(SessionTips {
root: root_tip,
mounts,
})
}
fn mount(&self, name: &str) -> Result<&MountedSource, Error> {
self.mounts
.iter()
.find(|mount| mount.name == name)
.ok_or_else(|| Error::Engine(format!("no source is mounted at {name:?}")))
}
fn session_target(&self, session: &Session, path: &str) -> (Option<String>, PathBuf, String) {
if let Some((first, rest)) = path.split_once('/')
&& !rest.is_empty()
&& self.mounts.iter().any(|mount| mount.name == first)
{
return (
Some(first.to_owned()),
session.working_copy.join(first),
rest.to_owned(),
);
}
(None, session.working_copy.clone(), path.to_owned())
}
fn gate_reacts_to_snapshot(
&mut self,
session: &Session,
new_snapshot: &str,
) -> Result<(), Error> {
let Some(request) = self.coordination.gated_request_for_session(session.id.0)? else {
return Ok(());
};
let id = RequestId(request.id);
match request.state {
RequestState::Open | RequestState::Approved | RequestState::Parked => {
if self.config()?.landing.dismiss_approvals_on_new_snapshots {
let dismissed = self.coordination.dismiss_approvals(request.id)?;
if dismissed > 0 {
self.append_session_entry(
&session.actor,
Act::ApprovalsDismissed,
session.id,
Some(format!("{id} {new_snapshot}")),
)?;
}
}
match request.state {
RequestState::Approved | RequestState::Parked => {
let _ = self.coordination.move_request_state(
request.id,
&[RequestState::Approved, RequestState::Parked],
RequestState::Open,
)?;
}
RequestState::Open
| RequestState::Landed
| RequestState::Rejected
| RequestState::Abandoned => {}
}
}
RequestState::Landed | RequestState::Rejected | RequestState::Abandoned => {}
}
Ok(())
}
fn gated_request(&mut self, id: RequestId) -> Result<RequestRow, Error> {
let Some(row) = self.coordination.request(id.0)? else {
return Err(Error::RequestNotFound(id.to_string()));
};
match row.state {
RequestState::Open | RequestState::Approved => Ok(row),
RequestState::Parked => Err(Error::RequestParked(id.to_string())),
RequestState::Landed | RequestState::Rejected | RequestState::Abandoned => {
Err(Error::RequestClosed {
id: id.to_string(),
state: row.state.to_string(),
})
}
}
}
fn open_session_only(&mut self, id: SessionId) -> Result<Session, Error> {
let session = self.session(id)?;
match session.state {
SessionState::Open => Ok(session),
SessionState::Landed | SessionState::Abandoned => Err(Error::SessionClosed {
id: id.to_string(),
state: session.state.to_string(),
}),
}
}
fn session_from(&self, row: SessionRow) -> Result<Session, Error> {
let id = SessionId(row.id);
let change_id = row.change_id.ok_or_else(|| {
Error::Engine(format!("session {id} has no change; its bootstrap failed"))
})?;
let mut changes = vec![SourceChange {
source: None,
change_id: change_id.clone(),
}];
for (source, mount_change) in self.coordination.session_source_changes(row.id)? {
changes.push(SourceChange {
source: Some(source),
change_id: mount_change,
});
}
Ok(Session {
id,
actor: Actor {
name: row.actor_name,
kind: row.actor_kind,
},
state: row.state,
change_id,
changes,
working_copy: self.session_root(id),
instruction_summary: row.instruction_summary,
instruction_run_ref: row.instruction_run_ref,
opened_at_ms: row.opened_at_ms,
})
}
fn request_from(&self, row: RequestRow) -> Result<LandingRequest, Error> {
let approvals = self
.coordination
.live_approvals(row.id)?
.into_iter()
.map(|approval| Approval {
actor: Actor {
name: approval.actor_name,
kind: approval.actor_kind,
},
snapshot: approval.snapshot_id,
at_ms: approval.at_ms,
})
.collect();
Ok(LandingRequest {
id: RequestId(row.id),
session_id: SessionId(row.session_id),
requester: Actor {
name: row.requester_name,
kind: row.requester_kind,
},
state: row.state,
approvals,
created_at_ms: row.created_at_ms,
})
}
fn session_root(&self, id: SessionId) -> PathBuf {
self.root
.join(CONTROL_DIR)
.join(SESSIONS_DIR)
.join(id.to_string())
}
fn config(&self) -> Result<WorkspaceConfig, Error> {
read_workspace_config(&self.root.join(CONTROL_DIR))
}
fn append_session_entry(
&self,
actor: &Actor,
act: Act,
session: SessionId,
reference: Option<String>,
) -> Result<(), Error> {
self.journal.append(&JournalEntry {
at_ms: now_ms()?,
actor_name: actor.name.clone(),
actor_kind: actor.kind,
act,
session: Some(session.to_string()),
instruction_summary: None,
instruction_run_ref: None,
instruction_verbatim: None,
reference,
})
}
fn project_for_read(&self, package: &dyn FormatPackage, bytes: &[u8]) -> Result<String, Error> {
let blob = FileBlob {
id: crate::projection::content_id(bytes),
bytes: bytes.to_vec(),
};
if let Some(text) = self.projections.read(package.id(), &blob) {
return Ok(text);
}
match catch_unwind(AssertUnwindSafe(|| package.project(&blob.bytes))) {
Ok(Ok(projection)) => {
let _ = self
.projections
.store(package.id(), &blob, &projection.text);
Ok(projection.text)
}
Ok(Err(error)) => Err(Error::PackageFailed {
package: package.id().to_string(),
reason: error.to_string(),
}),
Err(_) => Err(Error::PackageFailed {
package: package.id().to_string(),
reason: "the package panicked during projection".to_owned(),
}),
}
}
fn raised(
&self,
engine: &Engine,
diff: Diff,
sides: &DiffSides,
mount: Option<&str>,
) -> Result<Diff, Error> {
let mut deltas = Vec::new();
for delta in diff.deltas {
deltas.extend(self.raise(engine, delta, sides, mount)?);
}
Ok(Diff { deltas })
}
fn raise(
&self,
engine: &Engine,
delta: Delta,
sides: &DiffSides,
mount: Option<&str>,
) -> Result<Vec<Delta>, Error> {
let raw = delta.address.as_str().to_owned();
let mut delta = delta;
if let Some(mount) = mount {
delta.address = Address::new(format!("{mount}/{raw}"));
}
if delta.kind != DeltaKind::Changed {
return Ok(vec![delta]);
}
let (before, after) = match engine.read_file_sides(sides, &raw)? {
(Side::Blob(before), Side::Blob(after)) => (before, after),
(Side::TooLarge, _) | (_, Side::TooLarge) => {
self.file_too_large(delta.address.as_str())?;
return Ok(vec![delta]);
}
(Side::Absent, _) | (_, Side::Absent) => return Ok(vec![delta]),
};
if let Some(package) = self.detected(delta.address.as_str(), &after.bytes)? {
let projections = (
self.projection(package, delta.address.as_str(), &before)?,
self.projection(package, delta.address.as_str(), &after)?,
);
let (Some(projected_before), Some(projected_after)) = projections else {
return Ok(vec![delta]);
};
let raised = delta.at_text_rung(
diff_lines(&projected_before, &projected_after),
Some(package.id()),
);
return self.enriched(raised, package, &before, &after);
}
match (as_text(&before.bytes), as_text(&after.bytes)) {
(Some(before), Some(after)) => {
Ok(vec![delta.at_text_rung(diff_lines(before, after), None)])
}
_ => Ok(vec![delta]),
}
}
fn enriched(
&self,
raised: Delta,
package: &dyn FormatPackage,
before: &FileBlob,
after: &FileBlob,
) -> Result<Vec<Delta>, Error> {
let rich = match catch_unwind(AssertUnwindSafe(|| {
package.diff(&before.bytes, &after.bytes)
})) {
Ok(None) => return Ok(vec![raised]),
Ok(Some(Ok(rich))) => rich,
Ok(Some(Err(error))) => {
self.differ_failed(raised.address.as_str(), package.id(), &error.to_string())?;
return Ok(vec![raised]);
}
Err(_) => {
self.differ_failed(
raised.address.as_str(),
package.id(),
"the package panicked during diffing",
)?;
return Ok(vec![raised]);
}
};
if rich.is_empty() {
return Ok(vec![raised]);
}
let path = raised.address.as_str().to_owned();
let mut deltas = vec![Delta {
fidelity: Fidelity::Rich,
..raised
}];
deltas.extend(rich.into_iter().map(|delta| Delta {
address: Address::new(format!("{path} > {}", delta.address.as_str())),
..delta
}));
Ok(deltas)
}
fn detected(&self, address: &str, bytes: &[u8]) -> Result<Option<&dyn FormatPackage>, Error> {
if let Ok(package) = catch_unwind(AssertUnwindSafe(|| {
detect_package(&self.packages, address, bytes)
})) {
Ok(package)
} else {
self.package_failed(address, None, "a package panicked during detection")?;
Ok(None)
}
}
fn projection(
&self,
package: &dyn FormatPackage,
address: &str,
blob: &FileBlob,
) -> Result<Option<String>, Error> {
if let Some(text) = self.projections.read(package.id(), blob) {
return Ok(Some(text));
}
match catch_unwind(AssertUnwindSafe(|| package.project(&blob.bytes))) {
Ok(Ok(projection)) => {
let _ = self.projections.store(package.id(), blob, &projection.text);
Ok(Some(projection.text))
}
Ok(Err(error)) => {
self.package_failed(address, Some(package.id()), &error.to_string())?;
Ok(None)
}
Err(_) => {
self.package_failed(
address,
Some(package.id()),
"the package panicked during projection",
)?;
Ok(None)
}
}
}
fn package_failed(
&self,
address: &str,
package: Option<PackageId>,
reason: &str,
) -> Result<(), Error> {
let reference = match package {
Some(id) => format!("{address} {id} fell_back_to=binary: {reason}"),
None => format!("{address} fell_back_to=binary: {reason}"),
};
let entry = self.entry(Act::PackageFailed, Some(reference))?;
self.journal.append(&entry)
}
fn differ_failed(&self, address: &str, package: PackageId, reason: &str) -> Result<(), Error> {
let reference = format!("{address} {package} fell_back_to=text: {reason}");
let entry = self.entry(Act::PackageFailed, Some(reference))?;
self.journal.append(&entry)
}
fn file_too_large(&self, address: &str) -> Result<(), Error> {
let reference = format!(
"{address} exceeds the {LADDER_FILE_SIZE_MAX}-byte ladder cap; kept at the binary rung"
);
let entry = self.entry(Act::FileTooLarge, Some(reference))?;
self.journal.append(&entry)
}
fn entry(&self, act: Act, reference: Option<String>) -> Result<JournalEntry, Error> {
Ok(JournalEntry {
at_ms: now_ms()?,
actor_name: self.actor.name.clone(),
actor_kind: self.actor.kind,
act,
session: None,
instruction_summary: None,
instruction_run_ref: None,
instruction_verbatim: None,
reference,
})
}
fn mount_boundary(&self) -> Vec<String> {
self.mounts.iter().map(|mount| mount.name.clone()).collect()
}
}
fn builtin_packages() -> Vec<Box<dyn FormatPackage>> {
vec![Box::new(DocxPackage)]
}
struct SessionTips {
root: String,
mounts: Vec<(String, String)>,
}
impl SessionTips {
fn tip_of(&self, source: Option<&str>) -> String {
match source {
None => self.root.clone(),
Some(name) => self
.mounts
.iter()
.find(|(mount, _)| mount == name)
.map_or_else(|| self.root.clone(), |(_, tip)| tip.clone()),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SourceSnapshot {
pub source: Option<String>,
pub snapshot: Snapshot,
}
fn mount_names(config: &WorkspaceConfig) -> Vec<String> {
let mut names: Vec<String> = config
.sources
.iter()
.filter(|source| source.mount != ROOT_MOUNT)
.map(|source| source.mount.clone())
.collect();
names.sort();
names
}
fn valid_mount_name(name: &str) -> Result<(), Error> {
let flat = !name.is_empty()
&& name != "."
&& name != ".."
&& !name.contains('/')
&& !name.contains('\\');
if !flat || SKIP_NAMES.contains(&name) {
return Err(Error::Config(format!(
"mount name {name:?} must be one path component outside the engine's internals"
)));
}
Ok(())
}
fn workspace_name(root: &Path) -> String {
match root.file_name().and_then(|name| name.to_str()) {
Some(name) => name.to_owned(),
None => "workspace".to_owned(),
}
}
fn enclosing_workspace(root: &Path) -> Option<PathBuf> {
let mut current = root.parent();
while let Some(dir) = current {
if dir.join(CONTROL_DIR).exists() {
return Some(dir.to_path_buf());
}
current = dir.parent();
}
None
}
fn folder_fingerprint(folder: &Path) -> Result<String, Error> {
let mut hasher = Sha256::new();
hash_folder(&mut hasher, folder)?;
Ok(format!("{:x}", hasher.finalize()))
}
fn hash_folder(hasher: &mut Sha256, root: &Path) -> Result<(), Error> {
let mut files: Vec<(String, PathBuf, bool)> = Vec::new();
let mut pending = vec![root.to_path_buf()];
while let Some(dir) = pending.pop() {
for entry in fs::read_dir(&dir)? {
let entry = entry?;
let name = entry.file_name();
let Some(name) = name.to_str() else {
return Err(Error::Engine(format!(
"cannot fingerprint a non-utf8 name at {}",
entry.path().display()
)));
};
if SKIP_NAMES.contains(&name) {
continue;
}
let path = entry.path();
let file_type = entry.file_type()?;
if file_type.is_dir() {
pending.push(path);
} else {
let rel = path
.strip_prefix(root)
.map_err(engine_err)?
.to_string_lossy()
.into_owned();
files.push((rel, path, file_type.is_symlink()));
}
}
}
files.sort();
for (rel, path, is_symlink) in &files {
hasher.update(rel.as_bytes());
if *is_symlink {
hasher.update([1]);
hasher.update(fs::read_link(path)?.to_string_lossy().as_bytes());
} else {
hasher.update([2]);
hasher.update(fs::read(path)?);
}
hasher.update([0]);
}
Ok(())
}
fn adopted_branch(source: &Path) -> Result<Option<String>, Error> {
let head = fs::read_to_string(source.join(".git").join("HEAD"))?;
Ok(head
.trim()
.strip_prefix("ref: refs/heads/")
.map(str::to_owned))
}
fn mount_refusals(name: &str, config: &WorkspaceConfig, mount_dir: &Path) -> Result<(), Error> {
valid_mount_name(name)?;
if config.sources.iter().any(|s| s.mount == name) {
return Err(Error::AlreadyAttached);
}
if mount_dir.exists() {
return Err(Error::Config(format!(
"mount {name:?} collides with existing workspace content"
)));
}
Ok(())
}
fn folder_uses_lfs(folder: &Path) -> Result<bool, Error> {
let gitattributes = folder.join(".gitattributes");
if !gitattributes.is_file() {
return Ok(false);
}
let text = fs::read_to_string(&gitattributes)?;
Ok(text.contains("filter=lfs"))
}
fn copy_tree(source: &Path, target: &Path, skips: &[&str]) -> Result<(), Error> {
let mut pending = vec![(source.to_path_buf(), target.to_path_buf())];
while let Some((from_dir, to_dir)) = pending.pop() {
for entry in fs::read_dir(&from_dir)? {
let entry = entry?;
let name = entry.file_name();
if skips.iter().any(|skip| name == **skip) {
continue;
}
let from = entry.path();
let to = to_dir.join(&name);
if from.is_dir() {
fs::create_dir_all(&to)?;
pending.push((from, to));
} else {
fs::copy(&from, &to)?;
}
}
}
Ok(())
}
fn session_file(working_copy: &Path, path: &str) -> Result<PathBuf, Error> {
let relative = Path::new(path);
let stays_inside = relative.components().all(|component| match component {
Component::Normal(_) | Component::CurDir => true,
Component::ParentDir | Component::RootDir | Component::Prefix(_) => false,
});
if path.is_empty() || !stays_inside {
return Err(Error::PathOutsideWorkingCopy(path.to_owned()));
}
Ok(working_copy.join(relative))
}
fn land_hold_ms() -> Result<Option<u64>, Error> {
match env::var("ATELIER_LAND_HOLD_MS") {
Ok(value) => value.parse().map(Some).map_err(config_err),
Err(VarError::NotPresent) => Ok(None),
Err(error @ VarError::NotUnicode(_)) => Err(config_err(error)),
}
}
fn now_ms() -> Result<i64, Error> {
let elapsed = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(config_err)?;
i64::try_from(elapsed.as_millis()).map_err(config_err)
}