use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use anyhow::{bail, Context, Result};
use serde_json::Value;
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
pub struct Scope {
pub name: String,
pub recipients: Vec<String>,
#[serde(default)]
pub projects: BTreeMap<String, String>,
}
impl Scope {
#[must_use]
pub fn for_project(&self, project: &str) -> String {
self.projects
.get(project)
.cloned()
.unwrap_or_else(|| self.name.clone())
}
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
struct Local {
#[serde(default)]
default_scope: Option<String>,
#[serde(default)]
repos: Vec<String>,
}
fn local() -> Local {
std::fs::read_to_string(config_dir().join("sync.toml"))
.ok()
.and_then(|t| toml::from_str::<Local>(&t).ok())
.unwrap_or_default()
}
fn expand(path: &str) -> PathBuf {
match path.strip_prefix("~/") {
Some(rest) => std::env::var_os("HOME")
.map_or_else(|| PathBuf::from(path), |h| PathBuf::from(h).join(rest)),
None => PathBuf::from(path),
}
}
#[must_use]
pub fn sender_entity(host: &str) -> String {
format!("sync:{host}")
}
#[must_use]
pub fn host() -> String {
std::fs::read_to_string("/proc/sys/kernel/hostname")
.map(|h| h.trim().to_string())
.ok()
.filter(|h| !h.is_empty())
.unwrap_or_else(|| "host".into())
}
fn config_dir() -> PathBuf {
std::env::var_os("XDG_CONFIG_HOME")
.map(PathBuf::from)
.or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".config")))
.unwrap_or_else(|| PathBuf::from(".config"))
.join("ljos")
}
fn state_dir() -> PathBuf {
std::env::var_os("XDG_STATE_HOME")
.map(PathBuf::from)
.or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
.unwrap_or_else(|| PathBuf::from(".local/state"))
.join("ljos")
.join("sync")
}
#[must_use]
pub fn identity_path() -> PathBuf {
config_dir().join("age.key")
}
pub fn public_key() -> Result<String> {
let key = identity_path();
if !key.exists() {
if let Some(dir) = key.parent() {
std::fs::create_dir_all(dir)?;
}
let made = Command::new("age-keygen")
.arg("-o")
.arg(&key)
.stdin(Stdio::null())
.output()
.context("age-keygen not on PATH; install age")?;
if !made.status.success() {
bail!(
"age-keygen: {}",
String::from_utf8_lossy(&made.stderr).trim()
);
}
}
let out = Command::new("age-keygen")
.arg("-y")
.arg(&key)
.output()
.context("age-keygen not on PATH; install age")?;
if !out.status.success() {
bail!(
"age-keygen -y: {}",
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(String::from_utf8_lossy(&out.stdout).trim().to_string())
}
pub fn scope_of_repo(root: &Path) -> Result<Option<Scope>> {
let path = root.join(".ljos").join("sync.toml");
match std::fs::read_to_string(&path) {
Ok(text) => Ok(Some(
toml::from_str(&text).with_context(|| path.display().to_string())?,
)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(e).with_context(|| path.display().to_string()),
}
}
fn default_scope(repo_scope: &str) -> String {
local()
.default_scope
.unwrap_or_else(|| repo_scope.to_string())
}
#[must_use]
pub fn atom_scope(atom: &Value, default: &str) -> String {
atom["entities"]
.as_array()
.into_iter()
.flatten()
.filter_map(Value::as_str)
.find_map(|e| e.strip_prefix("scope:"))
.map_or_else(|| default.to_string(), str::to_string)
}
#[must_use]
pub fn travelling(atom: &Value) -> Value {
let mut a = atom.clone();
if let Some(map) = a.as_object_mut() {
map.remove("embedding");
}
a
}
#[must_use]
pub fn is_own(atom: &Value) -> bool {
!atom["entities"]
.as_array()
.into_iter()
.flatten()
.filter_map(Value::as_str)
.any(|e| e.starts_with("sync:"))
}
#[must_use]
pub fn log_plaintext(atoms: &[Value], scope: &str, default: &str) -> String {
let mut mine: Vec<Value> = atoms
.iter()
.filter(|a| is_own(a) && atom_scope(a, default) == scope)
.map(travelling)
.collect();
mine.sort_by(|a, b| a["id"].as_str().cmp(&b["id"].as_str()));
mine.iter().map(|a| format!("{a}\n")).collect()
}
fn log_dir(root: &Path) -> PathBuf {
root.join(".ljos").join("atoms")
}
fn seal(plain: &str, recipients: &[String], out: &Path) -> Result<()> {
if recipients.is_empty() {
bail!("sync: the scope names no recipients; add this machine's `ljos sync --key` to .ljos/sync.toml");
}
let mut cmd = Command::new("age");
for r in recipients {
cmd.arg("-r").arg(r);
}
let tmp = out.with_extension("age.tmp");
let mut child = cmd
.arg("-o")
.arg(&tmp)
.stdin(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.context("age not on PATH; install age")?;
use std::io::Write;
child
.stdin
.take()
.context("age: no stdin")?
.write_all(plain.as_bytes())?;
let done = child.wait_with_output()?;
if !done.status.success() {
let _ = std::fs::remove_file(&tmp);
bail!("age: {}", String::from_utf8_lossy(&done.stderr).trim());
}
std::fs::rename(&tmp, out)?;
Ok(())
}
fn open(path: &Path, identity: &Path) -> Result<Option<String>> {
let out = Command::new("age")
.arg("-d")
.arg("-i")
.arg(identity)
.arg(path)
.stdin(Stdio::null())
.output()
.context("age not on PATH; install age")?;
if out.status.success() {
Ok(Some(String::from_utf8_lossy(&out.stdout).into_owned()))
} else if String::from_utf8_lossy(&out.stderr).contains("no identity matched") {
Ok(None)
} else {
bail!(
"age -d {}: {}",
path.display(),
String::from_utf8_lossy(&out.stderr).trim()
)
}
}
fn age_missing_line() -> String {
"sync: age is not installed, so the log was not sealed; install age to sync it\n".into()
}
fn age_missing(err: &anyhow::Error) -> bool {
err.chain().any(|cause| {
cause
.downcast_ref::<std::io::Error>()
.is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound)
})
}
pub fn export(root: &Path, scope: &Scope, atoms: &[Value]) -> Result<String> {
let host = host();
let plain = log_plaintext(atoms, &scope.name, &default_scope(&scope.name));
let count = plain.lines().count();
let dir = log_dir(root);
std::fs::create_dir_all(&dir)?;
let stamp = dir.join(format!("{host}.digest"));
let digest = format!(
"{} {}\n",
crate::work_id(&plain),
crate::work_id(&scope.recipients.join(","))
);
if std::fs::read_to_string(&stamp).is_ok_and(|d| d == digest) {
return Ok(format!(
"sync: {count} {} atoms unchanged since the last export\n",
scope.name
));
}
if !crate::on_path("age") {
return Ok(age_missing_line());
}
if let Err(err) = seal(
&plain,
&scope.recipients,
&dir.join(format!("{host}.jsonl.age")),
) {
if age_missing(&err) {
return Ok(age_missing_line());
}
return Err(err);
}
std::fs::write(&stamp, digest)?;
Ok(format!(
"sync: exported {count} {} atoms to .ljos/atoms/{host}.jsonl.age, sealed to {} recipient{}\n",
scope.name,
scope.recipients.len(),
if scope.recipients.len() == 1 { "" } else { "s" }
))
}
#[derive(Debug, Default, PartialEq)]
pub struct Incoming {
pub post: Vec<Value>,
pub retired: Vec<String>,
}
#[must_use]
pub fn incoming(
plain: &str,
host: &str,
before: &BTreeSet<String>,
held: &BTreeSet<String>,
) -> Incoming {
let mut post = Vec::new();
let mut now = BTreeSet::new();
for line in plain.lines().filter(|l| !l.trim().is_empty()) {
let Ok(mut atom) = serde_json::from_str::<Value>(line) else {
continue;
};
let text = atom["text"].as_str().unwrap_or("").to_string();
if text.is_empty() {
continue;
}
let id = crate::work_id(&text);
now.insert(id.clone());
if held.contains(&id) {
continue;
}
if let Some(map) = atom.as_object_mut() {
let mut entities: Vec<Value> = map
.get("entities")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let tag = sender_entity(host);
if !entities.iter().any(|e| e.as_str() == Some(tag.as_str())) {
entities.push(Value::String(tag));
}
map.insert("entities".into(), Value::Array(entities));
map.insert(
"origin".into(),
Value::String(crate::admit::ORIGIN_PEER.to_string()),
);
map.remove("id");
}
post.push(atom);
}
let retired = before.difference(&now).cloned().collect();
Incoming { post, retired }
}
fn taken_path(scope: &str, host: &str) -> PathBuf {
state_dir().join(scope).join(format!("{host}.taken"))
}
fn read_taken(scope: &str, host: &str) -> BTreeSet<String> {
std::fs::read_to_string(taken_path(scope, host))
.map(|t| t.lines().map(str::to_string).collect())
.unwrap_or_default()
}
fn write_taken(scope: &str, host: &str, plain: &str) -> Result<()> {
let path = taken_path(scope, host);
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir)?;
}
let ids: BTreeSet<String> = plain
.lines()
.filter_map(|l| serde_json::from_str::<Value>(l).ok())
.filter_map(|a| a["text"].as_str().map(crate::work_id))
.collect();
std::fs::write(path, ids.into_iter().map(|i| i + "\n").collect::<String>())?;
Ok(())
}
pub fn import(root: &Path, scope: &Scope) -> Result<String> {
let me = host();
let identity = identity_path();
if !identity.exists() {
return Ok("sync: no age identity on this machine; `ljos sync --key` makes one\n".into());
}
let dir = log_dir(root);
let Ok(entries) = std::fs::read_dir(&dir) else {
return Ok(format!("sync: no {} logs yet\n", scope.name));
};
let client = crate::pack()?;
let mut out = String::new();
let mut paths: Vec<PathBuf> = entries
.flatten()
.map(|e| e.path())
.filter(|p| p.to_string_lossy().ends_with(".jsonl.age"))
.collect();
paths.sort();
for path in paths {
let host = path
.file_name()
.and_then(|n| n.to_str())
.and_then(|n| n.strip_suffix(".jsonl.age"))
.unwrap_or("")
.to_string();
if host.is_empty() || host == me {
continue;
}
let Some(plain) = open(&path, &identity)? else {
out.push_str(&format!(
"sync: {host}'s {} log is not sealed to this machine\n",
scope.name
));
continue;
};
let workspace = client.workspace();
let tag = sender_entity(&host);
let held: BTreeSet<String> = crate::atoms_lean(&client, &workspace)?
.iter()
.filter(|a| {
a["entities"]
.as_array()
.into_iter()
.flatten()
.any(|e| e.as_str() == Some(tag.as_str()))
})
.filter_map(|a| a["text"].as_str().map(crate::work_id))
.collect();
let got = incoming(&plain, &host, &read_taken(&scope.name, &host), &held);
let (mut kept, mut refused) = (0usize, 0usize);
for mut atom in got.post {
if let Some(map) = atom.as_object_mut() {
map.insert("workspace".into(), Value::String(workspace.clone()));
}
match crate::admit::post_kept(&client, &atom) {
Ok(_) => kept += 1,
Err(_) => refused += 1,
}
}
let mut retired = 0usize;
if !got.retired.is_empty() {
for atom in crate::atoms_lean(&client, &workspace)? {
let text = atom["text"].as_str().unwrap_or("");
let from_host = atom["entities"]
.as_array()
.into_iter()
.flatten()
.any(|e| e.as_str() == Some(tag.as_str()));
if from_host && got.retired.contains(&crate::work_id(text)) {
if let Some(id) = atom["id"].as_str() {
if crate::packset_forget(id, None).is_ok() {
retired += 1;
}
}
}
}
}
write_taken(&scope.name, &host, &plain)?;
out.push_str(&format!(
"sync: {host}: {kept} {} atoms taken, {refused} refused, {retired} retired\n",
scope.name
));
}
Ok(out)
}
#[must_use]
pub fn scope_for_issue(issue: &str) -> Option<String> {
let hit = vissue_core::Layout::resolve(None, None)
.and_then(vissue_core::Router::load)
.and_then(|router| router.find_by_id(issue))
.ok()?;
let root = hit
.path
.ancestors()
.find(|d| d.join(".git").exists())?
.to_path_buf();
scope_of_repo(&root)
.ok()
.flatten()
.map(|s| s.for_project(&hit.project))
}
fn tracker_root() -> Result<PathBuf> {
let layout = vissue_core::Layout::resolve(None, None).map_err(anyhow::Error::from)?;
Ok(layout.root().to_path_buf())
}
fn git(root: &Path, args: &[&str]) -> std::io::Result<std::process::Output> {
Command::new("git")
.arg("-C")
.arg(root)
.args(args)
.stdin(Stdio::null())
.output()
}
fn commit_if_sealed(root: &Path, scope: &str, exported: &str) -> String {
let age = log_dir(root).join(format!("{}.jsonl.age", host()));
if exported.contains("age is not installed") && !age.is_file() {
return String::new();
}
commit_log(root, scope)
}
fn push_pending(root: &Path) -> String {
let remotes = git(root, &["remote"])
.ok()
.filter(|o| o.status.success())
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.unwrap_or_default();
if remotes.is_empty() || !commits_to_push(root) {
return String::new();
}
match crate::tracker_push::push_checkout(root) {
Ok(()) => "sync: pushed pending commits\n".into(),
Err(e) => format!("sync: push refused: {e}\n"),
}
}
fn commits_to_push(root: &Path) -> bool {
let ahead = |spec: &str| -> Option<u64> {
let out = git(root, &["rev-list", "--count", spec]).ok()?;
if !out.status.success() {
return None;
}
String::from_utf8_lossy(&out.stdout).trim().parse().ok()
};
if ahead("@{upstream}..HEAD").is_some_and(|n| n > 0) {
return true;
}
if git(root, &["rev-parse", "--abbrev-ref", "@{upstream}"]).is_ok_and(|o| o.status.success()) {
return false;
}
let branch = git(root, &["rev-parse", "--abbrev-ref", "HEAD"])
.ok()
.filter(|o| o.status.success())
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.unwrap_or_default();
if branch.is_empty() || branch == "HEAD" {
return git(root, &["rev-parse", "--verify", "HEAD"]).is_ok_and(|o| o.status.success());
}
let remote_ref = format!("origin/{branch}");
match ahead(&format!("{remote_ref}..HEAD")) {
Some(n) => n > 0,
None => git(root, &["rev-parse", "--verify", "HEAD"]).is_ok_and(|o| o.status.success()),
}
}
fn commit_log(root: &Path, scope: &str) -> String {
let host = host();
let files = [
format!(".ljos/atoms/{host}.jsonl.age"),
format!(".ljos/atoms/{host}.digest"),
];
let common = git(root, &["rev-parse", "--git-common-dir"])
.ok()
.filter(|o| o.status.success())
.map(|o| root.join(String::from_utf8_lossy(&o.stdout).trim()))
.unwrap_or_else(|| root.join(".git"));
let held = crate::CommitLock::acquire(&common.join("ljos-commit.lock"));
let mut add = vec!["add", "--"];
add.extend(files.iter().map(String::as_str));
if git(root, &add).map_or(true, |o| !o.status.success()) {
return "sync: could not stage the log\n".into();
}
let staged = git(
root,
&["diff", "--cached", "--quiet", "--", &files[0], &files[1]],
);
if staged.is_ok_and(|o| o.status.success()) {
drop(held);
return push_pending(root);
}
let message = format!("chore(sync): {host} {scope} atoms");
let mut commit = vec!["commit", "-q", "--only", "-m", message.as_str(), "--"];
commit.extend(files.iter().map(String::as_str));
match git(root, &commit) {
Ok(o) if o.status.success() => {}
Ok(o) => {
return format!(
"sync: commit refused: {}\n",
String::from_utf8_lossy(&o.stderr)
.lines()
.next()
.unwrap_or("")
)
}
Err(e) => return format!("sync: git: {e}\n"),
}
drop(held);
let remotes = git(root, &["remote"])
.ok()
.filter(|o| o.status.success())
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.unwrap_or_default();
if remotes.is_empty() {
return format!(
"sync: committed {message}; no remote is configured, so nothing was pushed. \
Set one with `git -C {} remote add origin URL` and `git push -u origin HEAD`.\n",
root.display()
);
}
match crate::tracker_push::push_checkout(root) {
Ok(()) => format!("sync: committed {message}; pushed\n"),
Err(e) => format!("sync: committed {message}; push refused: {e}\n"),
}
}
fn catch_up(root: &Path) -> String {
if git(root, &["fetch", "-q", "--all"]).map_or(true, |o| !o.status.success()) {
return "sync: fetch failed; reading the logs as they are\n".into();
}
let Some(up) = crate::tracker_upstream(root) else {
return String::new();
};
let mut refs = vec![up.clone()];
refs.extend(
crate::tracker_mirrors(root, &up)
.unwrap_or_default()
.into_iter()
.map(|(r, b)| format!("{r}/{b}")),
);
let mut stuck = Vec::new();
for r in refs {
let forward = git(root, &["merge", "-q", "--ff-only", &r]);
if forward.is_ok_and(|o| o.status.success()) {
continue;
}
let merged = git(root, &["merge", "-q", "--no-edit", &r]);
if !merged.is_ok_and(|o| o.status.success()) {
let _ = git(root, &["merge", "--abort"]);
stuck.push(r);
}
}
if stuck.is_empty() {
String::new()
} else {
format!(
"sync: could not merge {}; `ljos doctor` names the split\n",
stuck.join(", ")
)
}
}
pub fn sync_repo(pull_import: bool, export_log: bool) -> Result<String> {
sync_repos(pull_import, export_log, true)
}
pub fn sync_repo_in_passing(pull_import: bool, export_log: bool) -> Result<String> {
sync_repos(pull_import, export_log, false)
}
fn sync_repos(pull_import: bool, export_log: bool, asked: bool) -> Result<String> {
let listed = local().repos;
let roots: Vec<PathBuf> = if listed.is_empty() {
vec![tracker_root()?]
} else {
listed.iter().map(|r| expand(r)).collect()
};
let mut out = String::new();
for root in roots {
out.push_str(&sync_one(&root, pull_import, export_log, asked)?);
}
Ok(out)
}
fn sync_one(root: &Path, pull_import: bool, export_log: bool, asked: bool) -> Result<String> {
let root = root.to_path_buf();
let mut out = String::new();
if let Err(e) = crate::tracker_push::prepare_tracker(&root, false, true) {
out.push_str(&format!("sync: merge setup: {e}\n"));
}
let Some(scope) = scope_of_repo(&root)? else {
if !asked {
return Ok(String::new());
}
out.push_str(&format!(
"sync: {} names no scope; write .ljos/sync.toml with name and recipients (`ljos sync --key` prints this machine's)\n",
root.display()
));
return Ok(out);
};
if pull_import {
out.push_str(&catch_up(&root));
out.push_str(&import(&root, &scope)?);
}
if export_log {
let client = crate::pack()?;
let atoms = crate::atoms_lean(&client, &client.workspace())?;
let said = export(&root, &scope, &atoms)?;
out.push_str(&said);
out.push_str(&commit_if_sealed(&root, &scope.name, &said));
}
Ok(out)
}
#[must_use]
pub fn user_rule_trackers() -> Vec<PathBuf> {
if std::env::var("LJOS_USER_RULES_SYNC").ok().as_deref() == Some("off") {
return Vec::new();
}
if let Some(one) = std::env::var("LJOS_USER_RULES_TRACKER")
.ok()
.filter(|path| !path.trim().is_empty())
{
return vec![expand(one.trim())];
}
let listed = local().repos;
if !listed.is_empty() {
return listed.iter().map(|path| expand(path)).collect();
}
if let Some(root) = std::env::var_os("ISSUE_ROOT")
.or_else(|| std::env::var_os("VISSUE_ROOT"))
.filter(|root| !root.is_empty())
{
return vec![PathBuf::from(root)];
}
if let Some(root) = seat_tracker() {
return vec![root];
}
if let Ok(cwd) = std::env::current_dir() {
if cwd.join("vissue.toml").is_file() {
return vec![cwd];
}
}
Vec::new()
}
fn seat_tracker() -> Option<PathBuf> {
#[derive(serde::Deserialize)]
struct Seat {
root: Option<String>,
}
let path = std::env::var_os("VISSUE_CONFIG")
.filter(|raw| !raw.is_empty())
.map(PathBuf::from)
.or_else(|| Some(config_dir().parent()?.join("vissue").join("config.toml")))?;
let text = std::fs::read_to_string(path).ok()?;
let root = toml::from_str::<Seat>(&text).ok()?.root?;
let path = expand(root.trim());
path.is_dir().then_some(path)
}
fn machine_name(host: &str) -> String {
let clean: String = host
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.' {
c
} else {
'-'
}
})
.collect();
let clean = clean.trim_matches('.').to_string();
if clean.is_empty() {
"host".into()
} else {
clean
}
}
fn looks_like_git(root: &Path) -> bool {
let git = root.join(".git");
git.is_dir() || git.is_file()
}
fn repo_rows(root: &Path) -> Vec<crate::user_rules::UserRule> {
let dir = root.join(".ljos").join("user-rules");
let Ok(read) = std::fs::read_dir(&dir) else {
return Vec::new();
};
let mut paths: Vec<PathBuf> = read
.filter_map(|entry| entry.ok())
.map(|entry| entry.path())
.filter(|path| path.extension().is_some_and(|ext| ext == "json"))
.collect();
paths.sort();
let mut rows = Vec::new();
for path in paths {
let Ok(text) = std::fs::read_to_string(&path) else {
continue;
};
if let Ok(parsed) = crate::user_rules::parse_rules(&text) {
rows.extend(parsed);
}
}
rows
}
fn git_within(
root: &Path,
args: &[&str],
deadline: std::time::Instant,
) -> Result<std::process::Output> {
use std::os::unix::process::CommandExt;
let budget = deadline.saturating_duration_since(std::time::Instant::now());
if budget.is_zero() {
bail!(
"sync: user rules: no time left for git {}",
args.first().copied().unwrap_or("")
);
}
let mut cmd = Command::new("git");
cmd.arg("-C")
.arg(root)
.args(args)
.env("GIT_TERMINAL_PROMPT", "0")
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
unsafe {
cmd.pre_exec(|| {
libc::setpgid(0, 0);
Ok(())
});
}
let child = cmd.spawn().with_context(|| {
format!(
"sync: user rules: git {}",
args.first().copied().unwrap_or("")
)
})?;
let pid = child.id();
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = tx.send(child.wait_with_output());
});
match rx.recv_timeout(budget) {
Ok(Ok(out)) => Ok(out),
Ok(Err(err)) => Err(err).context("sync: user rules: git"),
Err(_) => {
unsafe {
libc::kill(-(pid as i32), libc::SIGKILL);
libc::kill(pid as i32, libc::SIGKILL);
}
bail!(
"sync: user rules: the tracker did not answer within {} ms",
budget.as_millis()
)
}
}
}
fn git_succeeded(out: &std::process::Output) -> bool {
out.status.success()
}
fn one_line(out: &std::process::Output) -> String {
let text = String::from_utf8_lossy(&out.stderr);
let text = if text.trim().is_empty() {
String::from_utf8_lossy(&out.stdout)
} else {
text
};
text.lines()
.find(|line| !line.trim().is_empty())
.unwrap_or("")
.trim()
.to_string()
}
fn integrate_remote(root: &Path, deadline: std::time::Instant) -> Result<()> {
let fetched = git_within(root, &["fetch", "-q", "--all"], deadline)?;
if !git_succeeded(&fetched) {
bail!("sync: user rules: fetch failed: {}", one_line(&fetched));
}
let Some(up) = crate::tracker_upstream(root) else {
return Ok(());
};
let forward = git_within(root, &["merge", "-q", "--ff-only", &up], deadline)?;
if git_succeeded(&forward) {
return Ok(());
}
let merged = git_within(root, &["merge", "-q", "--no-edit", &up], deadline)?;
if git_succeeded(&merged) {
return Ok(());
}
let _ = git_within(
root,
&["merge", "--abort"],
std::time::Instant::now() + std::time::Duration::from_secs(2),
);
bail!(
"sync: user rules: could not merge {up}: {}",
one_line(&merged)
)
}
fn commit_host(root: &Path, rel: &str, host: &str, deadline: std::time::Instant) -> Result<bool> {
let common = git(root, &["rev-parse", "--git-common-dir"])
.ok()
.filter(|out| out.status.success())
.map(|out| {
let rel_git = String::from_utf8_lossy(&out.stdout).trim().to_string();
let path = PathBuf::from(&rel_git);
if path.is_absolute() {
path
} else {
root.join(path)
}
})
.unwrap_or_else(|| root.join(".git"));
let held = crate::CommitLock::acquire(&common.join("ljos-commit.lock"));
let added = git_within(root, &["add", "--", rel], deadline)?;
if !git_succeeded(&added) {
bail!(
"sync: user rules: could not stage {rel}: {}",
one_line(&added)
);
}
let staged = git_within(root, &["diff", "--cached", "--quiet", "--", rel], deadline)?;
if git_succeeded(&staged) {
drop(held);
return Ok(false);
}
let message = format!("chore(sync): {host} user rules");
let committed = git_within(
root,
&["commit", "-q", "--only", "-m", &message, "--", rel],
deadline,
)?;
drop(held);
if git_succeeded(&committed) {
Ok(true)
} else {
bail!("sync: user rules: commit refused: {}", one_line(&committed))
}
}
fn push_host(root: &Path, deadline: std::time::Instant) -> Result<()> {
let upstream = git(root, &["rev-parse", "--abbrev-ref", "@{upstream}"])
.ok()
.is_some_and(|out| out.status.success());
let args: &[&str] = if upstream {
&["push", "-q"]
} else {
&["push", "-q", "-u", "origin", "HEAD"]
};
let pushed = git_within(root, args, deadline)?;
if git_succeeded(&pushed) {
return Ok(());
}
let said = format!(
"{}\n{}",
one_line(&pushed),
String::from_utf8_lossy(&pushed.stderr)
);
if !crate::tracker_push::rejected(&said) {
bail!("sync: user rules: push refused: {}", one_line(&pushed));
}
integrate_remote(root, deadline)?;
let again = git_within(root, args, deadline)?;
if git_succeeded(&again) {
Ok(())
} else {
bail!("sync: user rules: push refused: {}", one_line(&again))
}
}
pub fn exchange_user_rules(root: &Path, host: &str, budget: std::time::Duration) -> Result<String> {
if !root.is_dir() {
bail!("sync: user rules: {} is not a directory", root.display());
}
if !looks_like_git(root) {
bail!("sync: user rules: {} is not a git checkout", root.display());
}
let deadline = std::time::Instant::now() + budget;
let remote = git(root, &["remote"])
.ok()
.filter(|out| out.status.success())
.is_some_and(|out| !String::from_utf8_lossy(&out.stdout).trim().is_empty());
if remote {
integrate_remote(root, deadline)?;
} else {
let local_empty = crate::user_rules::read_all().is_ok_and(|rows| rows.is_empty());
if local_empty && repo_rows(root).is_empty() {
return Ok(String::new());
}
}
let mut rows = repo_rows(root);
rows.extend(crate::user_rules::read_all()?);
let merged = crate::user_rules::merge_rows(rows);
crate::user_rules::replace_stored(&merged)?;
let name = machine_name(host);
let rel = format!(".ljos/user-rules/{name}.json");
let path = root.join(&rel);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(&path, crate::user_rules::render_rules(&merged)?)?;
let committed = commit_host(root, &rel, &name, deadline)?;
let mut out = String::new();
if remote {
out.push_str("sync: user rules: fetched\n");
}
if committed || (remote && commits_to_push(root)) {
push_host(root, deadline)?;
out.push_str("sync: user rules: pushed\n");
}
let live = merged.iter().filter(|rule| !rule.deleted).count();
out.push_str(&format!("sync: user rules: {live} on this machine\n"));
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
fn atom(id: &str, text: &str, entities: &[&str]) -> Value {
serde_json::json!({"id": id, "text": text, "kind": "lesson",
"entities": entities, "embedding": [0.1, 0.2]})
}
#[test]
fn a_repository_with_no_scope_is_named_only_when_sync_is_asked_for() {
let _scratch = tempfile::tempdir().unwrap();
let root = _scratch.path().to_path_buf();
let asked = sync_one(&root, true, false, true).unwrap();
assert!(asked.contains("names no scope"), "{asked}");
assert_eq!(sync_one(&root, true, false, false).unwrap(), "");
assert_eq!(sync_one(&root, false, true, false).unwrap(), "");
let _ = std::fs::remove_dir(&root);
}
#[test]
fn a_log_with_no_remote_says_how_to_set_one() {
let _g = crate::test_env_lock();
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let git = |args: &[&str]| {
let out = std::process::Command::new("git")
.arg("-C")
.arg(root)
.args(args)
.output()
.unwrap();
assert!(
out.status.success(),
"git {args:?}: {}",
String::from_utf8_lossy(&out.stderr)
);
};
git(&["init", "-q"]);
git(&["config", "user.email", "seat@example.invalid"]);
git(&["config", "user.name", "seat"]);
git(&["config", "core.hooksPath", "/dev/null"]);
let atoms = root.join(".ljos/atoms");
std::fs::create_dir_all(&atoms).unwrap();
let name = host();
std::fs::write(atoms.join(format!("{name}.jsonl.age")), "sealed\n").unwrap();
std::fs::write(atoms.join(format!("{name}.digest")), "digest\n").unwrap();
let said = commit_log(root, "surf");
assert!(said.contains("no remote is configured"), "{said}");
assert!(said.contains("remote add origin"), "{said}");
assert!(said.contains("git push -u origin HEAD"), "{said}");
assert!(!said.contains("pushed to nothing"), "{said}");
assert_eq!(commit_log(root, "surf"), "");
}
#[test]
fn unchanged_atoms_still_push_a_pending_commit() {
let _g = crate::test_env_lock();
let dir = tempfile::tempdir().unwrap();
let remote = dir.path().join("remote.git");
let root = dir.path().join("local");
let git = |cwd: &std::path::Path, args: &[&str]| {
let out = std::process::Command::new("git")
.arg("-C")
.arg(cwd)
.args(args)
.output()
.unwrap();
assert!(
out.status.success(),
"git {args:?}: {}",
String::from_utf8_lossy(&out.stderr)
);
};
std::fs::create_dir_all(&remote).unwrap();
git(&remote, &["init", "-q", "--bare"]);
std::fs::create_dir_all(&root).unwrap();
git(&root, &["init", "-q"]);
git(&root, &["config", "user.email", "seat@example.invalid"]);
git(&root, &["config", "user.name", "seat"]);
git(&root, &["config", "core.hooksPath", "/dev/null"]);
let atoms = root.join(".ljos/atoms");
std::fs::create_dir_all(&atoms).unwrap();
let name = host();
std::fs::write(atoms.join(format!("{name}.jsonl.age")), "sealed\n").unwrap();
std::fs::write(atoms.join(format!("{name}.digest")), "digest\n").unwrap();
std::fs::write(root.join("README"), "seed\n").unwrap();
git(&root, &["add", "."]);
git(&root, &["commit", "-q", "-m", "seed"]);
git(
&root,
&["remote", "add", "origin", remote.to_str().unwrap()],
);
git(&root, &["push", "-q", "-u", "origin", "HEAD"]);
std::fs::write(root.join(".gitattributes"), "*.org merge=union\n").unwrap();
git(&root, &["add", ".gitattributes"]);
git(
&root,
&[
"commit",
"-q",
"-m",
"chore(tracker): union org files and re-export atoms on merge",
],
);
let said = commit_log(&root, "surf");
assert!(
said.contains("pushed pending commits"),
"an unchanged atom log must still push: {said}"
);
let remote_log = std::process::Command::new("git")
.arg("-C")
.arg(&remote)
.args(["log", "--format=%s"])
.output()
.unwrap();
let subjects = String::from_utf8_lossy(&remote_log.stdout);
assert!(
subjects.contains("union org files"),
"the pending commit was not pushed: {subjects}"
);
assert_eq!(commit_log(&root, "surf"), "");
}
#[test]
fn a_missing_age_skips_the_seal_without_an_os_error() {
let _g = crate::test_env_lock();
struct Restore(Option<std::ffi::OsString>);
impl Drop for Restore {
fn drop(&mut self) {
match self.0.take() {
Some(path) => unsafe { std::env::set_var("PATH", path) },
None => unsafe { std::env::remove_var("PATH") },
}
}
}
let _path = Restore(std::env::var_os("PATH"));
let empty = tempfile::tempdir().unwrap();
unsafe { std::env::set_var("PATH", empty.path()) };
let root = tempfile::tempdir().unwrap();
let scope: Scope =
toml::from_str("name = \"surf\"\nrecipients = [\"age1test\"]\n").unwrap();
let said = export(root.path(), &scope, &[atom("a", "kept", &[])]).unwrap();
assert!(said.contains("age is not installed"), "{said}");
assert!(said.contains("install age"), "{said}");
assert!(!said.contains("os error"), "{said}");
assert!(!said.to_lowercase().contains("not found"), "{said}");
assert!(!log_dir(root.path())
.join(format!("{}.jsonl.age", host()))
.is_file());
assert_eq!(commit_if_sealed(root.path(), "surf", &said), "");
}
#[test]
fn a_log_carries_own_atoms_of_its_scope_without_vectors() {
let atoms = vec![
atom("b", "second", &["seat:x"]),
atom("a", "first", &[]),
atom("c", "personal", &["scope:personal"]),
atom("d", "imported", &["sync:otherhost"]),
];
let plain = log_plaintext(&atoms, "surf", "surf");
let lines: Vec<Value> = plain
.lines()
.map(|l| serde_json::from_str(l).unwrap())
.collect();
assert_eq!(lines.len(), 2, "{plain}");
assert_eq!(lines[0]["id"], "a");
assert_eq!(lines[1]["id"], "b");
assert!(lines.iter().all(|a| a.get("embedding").is_none()));
assert_eq!(log_plaintext(&atoms, "personal", "surf").lines().count(), 1);
}
#[test]
fn a_project_named_in_the_scope_file_takes_its_own_scope() {
let scope: Scope = toml::from_str(
"name = \"personal\"\nrecipients = [\"age1x\"]\n[projects]\ntools = \"shared\"\n",
)
.unwrap();
assert_eq!(scope.for_project("tools"), "shared");
assert_eq!(scope.for_project("garden"), "personal");
let bare: Scope = toml::from_str("name = \"shared\"\nrecipients = []\n").unwrap();
assert!(bare.projects.is_empty());
}
#[test]
fn an_import_tags_the_sender_and_names_what_it_retired() {
let plain = format!(
"{}\n{}\n",
atom("a", "kept", &["seat:x"]),
atom("b", "new", &[])
);
let before: BTreeSet<String> = [crate::work_id("kept"), crate::work_id("dropped")].into();
let got = incoming(&plain, "rglat", &before, &BTreeSet::new());
assert_eq!(got.post.len(), 2);
let again = incoming(&plain, "rglat", &before, &[crate::work_id("kept")].into());
assert_eq!(
again.post.len(),
1,
"a text already held is not posted again"
);
assert_eq!(again.post[0]["text"], "new");
assert!(got.post.iter().all(|a| a.get("id").is_none()));
assert!(got.post.iter().all(|a| a["entities"]
.as_array()
.unwrap()
.iter()
.any(|e| e == "sync:rglat")));
assert!(got.post.iter().all(|a| a["origin"] == "peer"));
assert_eq!(got.retired, vec![crate::work_id("dropped")]);
let claimed = "{\"id\":\"c\",\"text\":\"claimed\",\"kind\":\"lesson\",\"origin\":\"user-declared\",\"entities\":[]}\n";
let overwrote = incoming(claimed, "rglat", &BTreeSet::new(), &BTreeSet::new());
assert_eq!(overwrote.post[0]["origin"], "peer");
}
#[test]
fn a_sealed_log_opens_for_a_recipient_and_not_for_another() {
let _g = crate::test_env_lock();
let dir = tempfile::tempdir().unwrap();
let key = |name: &str| {
let path = dir.path().join(name);
let made = Command::new("age-keygen")
.arg("-o")
.arg(&path)
.output()
.unwrap();
assert!(made.status.success(), "age-keygen must be installed");
let public = Command::new("age-keygen")
.arg("-y")
.arg(&path)
.output()
.unwrap();
(
path,
String::from_utf8_lossy(&public.stdout).trim().to_string(),
)
};
let (mine, my_public) = key("mine.key");
let (other, _) = key("other.key");
let out = dir.path().join("host.jsonl.age");
seal("{\"text\":\"x\"}\n", &[my_public], &out).unwrap();
assert_eq!(
open(&out, &mine).unwrap().as_deref(),
Some("{\"text\":\"x\"}\n")
);
assert_eq!(open(&out, &other).unwrap(), None);
assert!(seal("x", &[], &out).is_err(), "no recipients is refused");
}
#[test]
fn an_atoms_scope_is_its_entity_or_the_default() {
assert_eq!(
atom_scope(&atom("a", "t", &["scope:personal"]), "surf"),
"personal"
);
assert_eq!(atom_scope(&atom("a", "t", &["seat:x"]), "surf"), "surf");
}
}