use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use anyhow::Context as _;
use serde_json::{Value, json};
use crate::config::{Hook, ValidatedConfig};
use crate::manifest::{GenerationManifest, RepoEntry};
pub(crate) mod diff;
#[cfg(test)]
mod tests;
pub(crate) const FORMAT: u32 = 1;
pub(crate) const HOOK_FILE: &str = "hook.json";
pub(crate) const DIFF_FILE: &str = "xrpc-diff.jsonl";
const HOOKS_DIR: &str = "runtime/hooks";
const SUPERSEDE_WAIT: Duration = Duration::from_secs(2);
pub(crate) struct Activation<'a> {
pub(crate) generation_dir: &'a Path,
pub(crate) manifest: &'a GenerationManifest,
pub(crate) previous: Option<&'a GenerationManifest>,
pub(crate) rollback: bool,
pub(crate) activated_at_unix: u64,
}
pub(crate) fn after_activation(config: &ValidatedConfig, activation: &Activation) {
if config.activate.hooks.is_empty() {
return;
}
match write_hook(config, activation) {
Ok(path) => start_all(config, &activation.manifest.generation_id, &path, false),
Err(error) => tracing::warn!("hooks skipped: {error:#}"),
}
}
pub(crate) fn start_all(
config: &ValidatedConfig,
generation: &str,
hook_json: &Path,
// a deploy's re-run leaves a hook that already started for this generation as it is
changed_only: bool,
) {
for hook in &config.activate.hooks {
if changed_only && last_started(config, hook).as_ref() == Some(&started(hook, generation)) {
continue;
}
if let Err(error) = start(config, hook, generation, hook_json) {
tracing::warn!("hook {} skipped: {error:#}", hook.name);
}
}
}
pub(crate) fn write_hook(
config: &ValidatedConfig,
activation: &Activation,
) -> anyhow::Result<PathBuf> {
let dir = activation.generation_dir;
let started = Instant::now();
let new_maps = diff::read_maps(dir)?;
let previous = activation.previous.and_then(|manifest| {
let old_dir = config
.output_root
.join("generations")
.join(&manifest.generation_id);
let maps = diff::read_maps(&old_dir)
.inspect_err(|error| tracing::warn!("hooks: previous maps unreadable: {error:#}"))
.ok()?;
Some((manifest, old_dir, maps))
});
let lines = previous.as_ref().map(|(manifest, _, maps)| {
diff::diff(
&config.output_root,
(manifest, maps),
(activation.manifest, &new_maps),
)
});
let diff_path = dir.join(DIFF_FILE);
let mut diff_bytes = 0;
match &lines {
Some(lines) => {
let mut body = Vec::new();
for line in lines {
serde_json::to_writer(&mut body, line)?;
body.push(b'\n');
}
diff_bytes = body.len();
crate::cas::atomic_write(&diff_path, &body)?;
}
None => match std::fs::remove_file(&diff_path) {
Err(error) if error.kind() != std::io::ErrorKind::NotFound => return Err(error.into()),
_ => {}
},
}
let new_handles = owner_handles(&new_maps, dir);
let old_handles = previous
.as_ref()
.map(|(_, old_dir, maps)| owner_handles(maps, old_dir))
.unwrap_or_default();
let mut summary = summarize(
activation.previous,
activation.manifest,
lines.as_deref().unwrap_or_default(),
(&old_handles, &new_handles),
);
let branches = default_branches(&new_maps, dir);
for repo in &mut summary.repos {
let branch = repo["did"].as_str().and_then(|did| branches.get(did));
repo["default_branch"] = json!(branch);
}
let hook = json!({
"format": FORMAT,
"knot_static_version": env!("CARGO_PKG_VERSION"),
"generation": activation.manifest.generation_id,
"previous": activation.previous.map(|manifest| &manifest.generation_id),
"rollback": activation.rollback,
"activated_at": rfc3339(activation.activated_at_unix),
"knot": {
"did": config.knot_did.as_str(),
"hostname": config.hostname.as_str(),
"owner": config.owner.as_str(),
"base_url": format!("https://{}", config.hostname.as_str()),
},
"paths": {
"generation": dir,
"www": dir.join("www"),
"diff": lines.as_ref().map(|_| &diff_path),
},
"diff": lines.as_ref().map(|lines| json!({
"lines": lines.len(),
"bytes": diff_bytes,
"seconds": started.elapsed().as_secs_f64(),
})),
"trigger": summary.trigger,
"repos": summary.repos,
"removed_repos": summary.removed,
});
let path = dir.join(HOOK_FILE);
crate::cas::atomic_write(&path, &serde_json::to_vec_pretty(&hook)?)?;
Ok(path)
}
fn rfc3339(unix: u64) -> String {
chrono::DateTime::from_timestamp(unix as i64, 0)
.unwrap_or_default()
.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
}
fn owner_handles(maps: &[diff::MapBlock], dir: &Path) -> BTreeMap<String, String> {
let Some(docs) = maps.iter().find(|map| map.target == "$ks_soc_doc") else {
return BTreeMap::new();
};
docs.rows
.iter()
.filter_map(|(key, file)| {
let did = diff::decode(key)?;
let bytes = std::fs::read(dir.join("www/_atproto").join(file)).ok()?;
let doc: Value = serde_json::from_slice(&bytes).ok()?;
let handle = doc["handle"].as_str()?;
(did.starts_with("did:") && handle != "handle.invalid")
.then(|| (did, handle.to_string()))
})
.collect()
}
fn default_branches(maps: &[diff::MapBlock], dir: &Path) -> BTreeMap<String, String> {
let Some(records) = maps.iter().find(|map| map.target == "$ks_atp_rec") else {
return BTreeMap::new();
};
records
.rows
.iter()
.filter_map(|(key, file)| {
let token = key.strip_suffix("|org.tangled.git.refs|self|")?;
let bytes = std::fs::read(dir.join("www/_atproto").join(token).join(file)).ok()?;
let record: Value = serde_json::from_slice(&bytes).ok()?;
let did = record["uri"]
.as_str()?
.strip_prefix("at://")?
.split('/')
.next()?;
Some((
did.to_string(),
record["value"]["defaultBranch"].as_str()?.to_string(),
))
})
.collect()
}
pub(crate) struct Summary {
pub(crate) trigger: Value,
pub(crate) repos: Vec<Value>,
pub(crate) removed: Vec<String>,
}
fn name_of(entry: &RepoEntry) -> &str {
entry.name.as_deref().unwrap_or(&entry.rkey)
}
/// activation doesn't say what caused it, so the trigger comes from the changes.
pub(crate) fn summarize(
previous: Option<&GenerationManifest>,
manifest: &GenerationManifest,
lines: &[diff::Line],
(old_handles, new_handles): (&BTreeMap<String, String>, &BTreeMap<String, String>),
) -> Summary {
let before: BTreeMap<&str, &RepoEntry> = previous
.map(|m| m.repos.iter().map(|r| (r.repo_did.as_str(), r)).collect())
.unwrap_or_default();
let mut cobs: BTreeMap<&str, Vec<Value>> = BTreeMap::new();
let mut touched: BTreeSet<&str> = BTreeSet::new();
for line in lines {
let Some(repo) = line.repo() else { continue };
touched.insert(repo);
if line.nsid != Some(diff::RECORD) || line.params.contains_key("cid") {
continue;
}
let collection = line.params.get("collection").map_or("", String::as_str);
if collection == "org.tangled.git.refs" {
continue;
}
let op = match line.op {
"added" => "create",
"removed" => "delete",
_ => "update",
};
cobs.entry(repo).or_default().push(json!({
"collection": collection,
"rkey": line.params.get("rkey"),
"op": op,
}));
}
let mut kinds = BTreeSet::new();
let mut changed = Vec::new();
let repos = manifest
.repos
.iter()
.map(|entry| {
let did = entry.repo_did.as_str();
let handle = new_handles.get(&entry.owner);
let mut repo = json!({
"did": did,
"owner_did": entry.owner,
"owner_handle": handle,
"name": name_of(entry),
});
let Some(old) = before.get(did) else {
repo["change"] = "added".into();
kinds.insert("repos");
changed.push(did);
return repo;
};
let refs = moved_refs(old, entry);
let renamed = name_of(old) != name_of(entry)
|| old.owner != entry.owner
|| old_handles.get(&old.owner) != handle;
let updated = old.fingerprint != entry.fingerprint
|| !refs.is_empty()
|| touched.contains(did)
|| cobs.contains_key(did);
repo["change"] = match (renamed, updated) {
(true, _) => "renamed",
(false, true) => "updated",
(false, false) => "unchanged",
}
.into();
if renamed {
repo["renamed_from"] = json!({
"owner_did": old.owner,
"owner_handle": old_handles.get(&old.owner),
"name": name_of(old),
});
kinds.insert("identity");
}
if !refs.is_empty() {
kinds.insert("push");
repo["refs"] = refs.into();
}
if let Some(ops) = cobs.remove(did) {
kinds.insert("cob");
repo["cobs"] = ops.into();
}
if renamed || updated {
changed.push(did);
}
repo
})
.collect();
let now: BTreeSet<&str> = manifest.repos.iter().map(|r| r.repo_did.as_str()).collect();
let removed: Vec<String> = before
.keys()
.filter(|did| !now.contains(*did))
.map(|did| did.to_string())
.collect();
if !removed.is_empty() {
kinds.insert("repos");
}
let kind = match previous {
None => "initial",
Some(_) => ["push", "cob", "repos", "identity"]
.into_iter()
.find(|kind| kinds.contains(kind))
.unwrap_or("other"),
};
Summary {
trigger: json!({ "kind": kind, "repos": changed }),
repos,
removed,
}
}
fn moved_refs(old: &RepoEntry, new: &RepoEntry) -> Vec<Value> {
let before: BTreeMap<&str, &str> = old
.public_refs
.iter()
.map(|(n, o)| (n.as_str(), o.as_str()))
.collect();
let after: BTreeMap<&str, &str> = new
.public_refs
.iter()
.map(|(n, o)| (n.as_str(), o.as_str()))
.collect();
let names: BTreeSet<&str> = before.keys().chain(after.keys()).copied().collect();
names
.into_iter()
.filter(|name| before.get(name) != after.get(name))
.map(|name| json!({ "ref": name, "old": before.get(name), "new": after.get(name) }))
.collect()
}
fn lock_file(hook: &Hook) -> String {
format!("{HOOKS_DIR}/{}.lock", hook.name)
}
fn pid_file(hook: &Hook) -> String {
format!("{HOOKS_DIR}/{}.pid", hook.name)
}
fn started_file(hook: &Hook) -> String {
format!("{HOOKS_DIR}/{}.started.json", hook.name)
}
fn log_file(hook: &Hook) -> String {
format!("hooks-{}.log", hook.name)
}
/// the hook keeps the lock after this process exits, so the fd has to survive exec.
pub(crate) fn start(
config: &ValidatedConfig,
hook: &Hook,
generation: &str,
hook_json: &Path,
) -> anyhow::Result<()> {
use std::os::fd::AsRawFd as _;
use std::os::unix::process::CommandExt as _;
let root = &config.output_root;
std::fs::create_dir_all(root.join(HOOKS_DIR))?;
let lock = crate::output::lock_file(root, &lock_file(hook))?;
if lock.try_lock().is_err() {
supersede(root, hook, &lock)?;
}
let log = crate::git_ssh::open_log_sink(config, &log_file(hook));
let fd = lock.as_raw_fd();
let mut command = std::process::Command::new(&hook.command);
command
.args(&hook.args)
.arg(hook_json)
.stdin(std::process::Stdio::null())
.stdout(log.try_clone()?)
.stderr(log)
.process_group(0);
// SAFETY: fcntl is async-signal-safe, and this only changes the forked child's copy of the fd
unsafe {
command.pre_exec(move || {
let flags = libc::fcntl(fd, libc::F_GETFD);
if flags < 0 || libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC) < 0 {
return Err(std::io::Error::last_os_error());
}
Ok(())
});
}
let mut child = command
.spawn()
.with_context(|| format!("start hook {} ({})", hook.name, hook.command.display()))?;
crate::cas::atomic_write(
&root.join(pid_file(hook)),
child.id().to_string().as_bytes(),
)?;
std::thread::spawn(move || child.wait());
crate::cas::atomic_write(
&root.join(started_file(hook)),
&serde_json::to_vec(&started(hook, generation))?,
)?;
Ok(())
}
fn started(hook: &Hook, generation: &str) -> Value {
json!({ "command": hook.command, "args": hook.args, "generation": generation })
}
fn last_started(config: &ValidatedConfig, hook: &Hook) -> Option<Value> {
let bytes = std::fs::read(config.output_root.join(started_file(hook))).ok()?;
serde_json::from_slice(&bytes).ok()
}
/// a configured hook isn't the one its last run started, a deploy changed it
pub(crate) fn changed(config: &ValidatedConfig) -> bool {
config.activate.hooks.iter().any(|hook| {
let last = last_started(config, hook);
last.as_ref().map(|v| (&v["command"], &v["args"]))
!= Some((&json!(hook.command), &json!(hook.args)))
})
}
fn supersede(root: &Path, hook: &Hook, lock: &std::fs::File) -> anyhow::Result<()> {
let group = std::fs::read_to_string(root.join(pid_file(hook)))
.ok()
.and_then(|pid| pid.trim().parse::<libc::pid_t>().ok())
.filter(|&pid| pid > 1);
for signal in [libc::SIGTERM, libc::SIGKILL] {
if let Some(group) = group {
// SAFETY: kill has no memory effects
unsafe { libc::kill(-group, signal) };
}
let deadline = Instant::now() + SUPERSEDE_WAIT;
while Instant::now() < deadline {
if lock.try_lock().is_ok() {
return Ok(());
}
std::thread::sleep(Duration::from_millis(20));
}
}
anyhow::bail!(
"the previous {} hook still holds {}",
hook.name,
lock_file(hook)
)
}