use crate::config::Config;
use crate::db::Database;
use crate::records::{self, EdgeRecord, NodeRecord, RecordStore};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::path::Path;
pub const TOKEN_ENV: &str = "DECIDUOUS_MCP_TOKEN";
pub const FALLBACK_WORKSPACE: &str = "scratch";
#[derive(Debug, Deserialize, Serialize, Default, Clone)]
pub struct RemoteConfig {
#[serde(default)]
pub url: Option<String>,
#[serde(default)]
pub workspace: Option<String>,
}
impl RemoteConfig {
pub fn is_configured(&self) -> bool {
self.url.is_some()
}
}
pub fn credentials_path() -> Option<std::path::PathBuf> {
let base = std::env::var_os("XDG_CONFIG_HOME")
.map(std::path::PathBuf::from)
.or_else(|| {
std::env::var_os("HOME").map(|h| std::path::PathBuf::from(h).join(".config"))
})?;
Some(base.join("deciduous").join("credentials"))
}
pub fn token() -> Result<String, String> {
if let Ok(t) = std::env::var(TOKEN_ENV) {
if !t.trim().is_empty() {
return Ok(t.trim().to_string());
}
}
if let Some(path) = credentials_path() {
if let Ok(contents) = std::fs::read_to_string(&path) {
let t = contents.trim();
if !t.is_empty() {
return Ok(t.to_string());
}
}
}
Err(format!(
"no token for the shared graph.\n\n\
Store one (kept outside every repository, mode 0600):\n\n \
deciduous remote login\n\n\
or set {TOKEN_ENV} for a one-off."
))
}
pub fn store_token(token: &str) -> Result<std::path::PathBuf, String> {
let token = token.trim();
if token.is_empty() {
return Err("refusing to store an empty token".to_string());
}
let path = credentials_path().ok_or("cannot determine a config directory (is HOME set?)")?;
let dir = path.parent().ok_or("credentials path has no parent")?;
std::fs::create_dir_all(dir).map_err(|e| format!("creating {}: {e}", dir.display()))?;
let mut opts = std::fs::OpenOptions::new();
opts.write(true).create(true).truncate(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
opts.mode(0o600);
}
let mut file = opts
.open(&path)
.map_err(|e| format!("opening {}: {e}", path.display()))?;
use std::io::Write;
writeln!(file, "{token}").map_err(|e| format!("writing {}: {e}", path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
}
Ok(path)
}
pub fn forget_token() -> Result<bool, String> {
let Some(path) = credentials_path() else {
return Ok(false);
};
match std::fs::remove_file(&path) {
Ok(()) => Ok(true),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(e) => Err(format!("removing {}: {e}", path.display())),
}
}
pub fn repo_root(dir: &Path) -> Option<std::path::PathBuf> {
let git = |args: &[&str]| {
std::process::Command::new("git")
.arg("-C")
.arg(dir)
.args(args)
.output()
.ok()
.filter(|o| o.status.success())
.and_then(|o| String::from_utf8(o.stdout).ok())
};
let out = git(&[
"rev-parse",
"--path-format=absolute",
"--show-toplevel",
"--git-dir",
"--git-common-dir",
])?;
let mut lines = out.lines();
let (top, git_dir, common) = (lines.next()?, lines.next()?, lines.next()?);
if git_dir == common {
return Some(std::path::PathBuf::from(top));
}
let list = git(&["worktree", "list", "--porcelain"])?;
list.lines()
.find_map(|l| l.strip_prefix("worktree "))
.map(std::path::PathBuf::from)
}
pub fn workspace_for(dir: &Path) -> String {
match repo_root(dir) {
Some(path) => path
.file_name()
.map(|n| {
let n = n.to_string_lossy().to_lowercase();
n.strip_suffix(".git").map(str::to_string).unwrap_or(n)
})
.unwrap_or_else(|| FALLBACK_WORKSPACE.to_string()),
None => FALLBACK_WORKSPACE.to_string(),
}
}
pub fn repo_roots(dir: &Path) -> Option<Vec<String>> {
let git = |args: &[&str]| {
std::process::Command::new("git")
.arg("-C")
.arg(dir)
.args(args)
.output()
.ok()
};
let inside = git(&["rev-parse", "--is-shallow-repository"])
.filter(|o| o.status.success())
.and_then(|o| String::from_utf8(o.stdout).ok())?;
if inside.trim() == "true" {
return None;
}
let out = git(&["rev-list", "--max-parents=0", "HEAD"])?;
if !out.status.success() {
return Some(Vec::new());
}
let mut roots: Vec<String> = String::from_utf8_lossy(&out.stdout)
.lines()
.map(|l| l.trim().to_string())
.filter(|l| !l.is_empty())
.collect();
roots.sort();
roots.dedup();
Some(roots)
}
pub fn is_shallow(dir: &Path) -> bool {
std::process::Command::new("git")
.arg("-C")
.arg(dir)
.args(["rev-parse", "--is-shallow-repository"])
.output()
.ok()
.filter(|o| o.status.success())
.is_some_and(|o| String::from_utf8_lossy(&o.stdout).trim() == "true")
}
fn project_config(dir: &Path) -> Option<std::path::PathBuf> {
dir.ancestors()
.map(|d| d.join(".deciduous").join("config.toml"))
.find(|p| p.exists())
}
fn legacy_workspace(
dir: &Path,
url: &str,
token: &str,
derived: String,
deadline: Option<std::time::Instant>,
) -> Result<String, String> {
use colored::Colorize;
let Some(config) = project_config(dir) else {
return Ok(derived);
};
let ids = local_change_ids(&config.with_file_name("deciduous.db"));
let chosen = if ids.is_empty() {
derived.clone()
} else {
#[derive(Deserialize)]
struct Held {
name: String,
nodes: usize,
}
#[derive(Deserialize)]
struct Reply {
workspaces: Vec<Held>,
}
let reply = ureq::post(&format!("{url}/locate"))
.set("authorization", &format!("Bearer {token}"))
.timeout(within(std::time::Duration::from_secs(15), deadline))
.send_json(serde_json::json!({ "change_ids": ids }))
.map_err(describe)
.and_then(|r| {
r.into_json::<Reply>()
.map_err(|e| format!("the server's /locate answer was not a list: {e}"))
});
let held = match reply {
Ok(r) => r.workspaces,
Err(_) => return Ok(derived),
};
match held.as_slice() {
[] => derived.clone(),
_ if held.iter().any(|h| h.name == derived) => derived.clone(),
[one] => one.name.clone(),
many => {
return Err(format!(
"this project's config names no workspace (it was written by 1.0.7, which used the \
directory name), and this project's nodes are in more than one workspace on {url}: {}.\n\n\
Record the one it should write to:\n\n deciduous remote init {url} --workspace <name>",
many.iter()
.map(|h| format!("{} ({} nodes)", h.name, h.nodes))
.collect::<Vec<_>>()
.join(", ")
))
}
}
};
let Ok(text) = std::fs::read_to_string(&config) else {
return Ok(chosen);
};
let Ok(mut doc) = text.parse::<toml_edit::DocumentMut>() else {
return Ok(chosen);
};
let same_url = doc
.get("remote")
.and_then(|r| r.get("url"))
.and_then(|u| u.as_str())
.is_some_and(|u| u.trim_end_matches('/') == url);
if !same_url || doc.get("remote").and_then(|r| r.get("workspace")).is_some() {
return Ok(chosen);
}
doc["remote"]["workspace"] = toml_edit::value(chosen.as_str());
let why = if chosen == derived {
String::new()
} else {
format!(
" (the server holds this project's nodes there: the name this directory had when 1.0.7 wrote them, not \"{derived}\")"
)
};
match std::fs::write(&config, doc.to_string()) {
Ok(()) => eprintln!(
"{} recorded workspace = \"{chosen}\"{why} in {}, so renaming or cloning this \
repository keeps writing to the same graph. Commit that file.",
"Note:".yellow(),
config.display()
),
Err(e) => eprintln!(
"{} could not record workspace = \"{chosen}\" in {}: {e}. \
Until it is recorded, renaming this directory changes which graph it writes to.",
"Warning:".yellow(),
config.display()
),
}
Ok(chosen)
}
fn local_change_ids(db: &Path) -> Vec<String> {
let Ok(conn) =
rusqlite::Connection::open_with_flags(db, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)
else {
return Vec::new();
};
let Ok(mut stmt) =
conn.prepare("SELECT change_id FROM decision_nodes WHERE change_id IS NOT NULL ORDER BY id DESC LIMIT 1000")
else {
return Vec::new();
};
stmt.query_map([], |r| r.get::<_, String>(0))
.map(|rows| rows.flatten().collect())
.unwrap_or_default()
}
#[derive(Debug)]
pub struct Remote {
pub url: String,
pub workspace: String,
token: String,
pub repo_roots: Option<Vec<String>>,
pub shallow: bool,
ops_timeout: std::time::Duration,
deadline: Option<std::time::Instant>,
}
fn within(limit: std::time::Duration, deadline: Option<std::time::Instant>) -> std::time::Duration {
let left = deadline.map_or(limit, |d| {
d.saturating_duration_since(std::time::Instant::now())
});
limit.min(left).max(std::time::Duration::from_millis(1))
}
const OPS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);
pub const OPS_TIMEOUT_ENV: &str = "DECIDUOUS_OPS_TIMEOUT";
fn ops_timeout() -> Result<std::time::Duration, String> {
match std::env::var(OPS_TIMEOUT_ENV) {
Err(_) => Ok(OPS_TIMEOUT),
Ok(v) => match v.trim().parse::<u64>() {
Ok(n) if n > 0 => Ok(std::time::Duration::from_secs(n)),
_ => Err(format!(
"{OPS_TIMEOUT_ENV}={v:?} is not a number of seconds above zero"
)),
},
}
}
pub const QUICK_OPS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
pub fn config_at(data_dir: &Path) -> Result<Config, String> {
let mut config = own_config_at(data_dir)?;
if !config.remote.is_configured() {
if let Some(inherited) = inherited_remote(data_dir) {
config.remote = inherited.remote;
}
}
Ok(config)
}
fn own_config_at(data_dir: &Path) -> Result<Config, String> {
let path = data_dir.join("config.toml");
match std::fs::read_to_string(&path) {
Ok(text) => toml::from_str(&text).map_err(|e| format!("{}: {e}", path.display())),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Config::default()),
Err(e) => Err(format!("{}: {e}", path.display())),
}
}
#[derive(Debug, Clone)]
pub struct InheritedRemote {
pub remote: RemoteConfig,
pub data_dir: std::path::PathBuf,
}
pub fn inherited_remote(data_dir: &Path) -> Option<InheritedRemote> {
if own_config_at(data_dir).ok()?.remote.is_configured() {
return None;
}
let data_dir = std::path::absolute(data_dir).ok()?;
let project = data_dir.parent()?;
let top = project.ancestors().find(|d| d.join(".git").exists())?;
if !top.join(".git").is_file() {
return None;
}
let main = repo_root(project)?;
let same = |a: &Path, b: &Path| match (a.canonicalize(), b.canonicalize()) {
(Ok(a), Ok(b)) => a == b,
_ => a == b,
};
if same(&main, top) {
return None;
}
let main_data_dir = main.join(data_dir.strip_prefix(top).ok()?);
if same(&main_data_dir, &data_dir) {
return None;
}
let remote = own_config_at(&main_data_dir).ok()?.remote;
remote.is_configured().then_some(InheritedRemote {
remote,
data_dir: main_data_dir,
})
}
impl Remote {
pub fn for_data_dir(data_dir: &Path) -> Result<Self, String> {
Self::for_data_dir_by(data_dir, None)
}
pub fn for_replay_after_write(data_dir: &Path) -> Result<Self, String> {
let deadline = std::time::Instant::now() + QUICK_OPS_TIMEOUT;
let mut r = Self::for_data_dir_by(data_dir, Some(deadline))?;
r.ops_timeout = QUICK_OPS_TIMEOUT;
r.deadline = Some(deadline);
Ok(r)
}
fn for_data_dir_by(
data_dir: &Path,
deadline: Option<std::time::Instant>,
) -> Result<Self, String> {
let mut config = own_config_at(data_dir)?;
let data_dir = std::path::absolute(data_dir).unwrap_or_else(|_| data_dir.to_path_buf());
let project = data_dir.parent().unwrap_or(&data_dir);
let inherited = (!config.remote.is_configured())
.then(|| inherited_remote(&data_dir))
.flatten();
let Some(inherited) = inherited else {
return Self::resolve_by(&config, project, deadline);
};
config.remote = inherited.remote;
let main = inherited.data_dir.parent().unwrap_or(&inherited.data_dir);
let mut remote = Self::resolve_by(&config, main, deadline)?;
remote.repo_roots = repo_roots(project);
remote.shallow = is_shallow(project);
Ok(remote)
}
pub fn resolve(config: &Config, dir: &Path) -> Result<Self, String> {
Self::resolve_by(config, dir, None)
}
fn resolve_by(
config: &Config,
dir: &Path,
deadline: Option<std::time::Instant>,
) -> Result<Self, String> {
let url = config.remote.url.clone().ok_or_else(|| {
"no remote configured for this project.\n\nRun:\n\n deciduous remote init <url>"
.to_string()
})?;
let url = url.trim_end_matches('/').to_string();
let token = token()?;
let workspace = match &config.remote.workspace {
Some(ws) => ws.clone(),
None => legacy_workspace(dir, &url, &token, workspace_for(dir), deadline)?,
};
Ok(Self {
url,
workspace,
token,
repo_roots: repo_roots(dir),
shallow: is_shallow(dir),
ops_timeout: ops_timeout()?,
deadline: None,
})
}
pub fn events_url(&self) -> String {
let ws_base = if let Some(rest) = self.url.strip_prefix("https://") {
format!("wss://{rest}")
} else if let Some(rest) = self.url.strip_prefix("http://") {
format!("ws://{rest}")
} else {
self.url.clone()
};
format!(
"{ws_base}/events?workspace={}&token={}",
urlencode(&self.workspace),
urlencode(&self.token)
)
}
fn get(&self, path: &str) -> ureq::Request {
let req = ureq::get(&format!("{}{}", self.url, path))
.set("authorization", &format!("Bearer {}", self.token));
match &self.repo_roots {
Some(roots) => req.set("x-deciduous-repo-roots", &roots.join(",")),
None => req,
}
}
fn post(&self, path: &str) -> ureq::Request {
ureq::post(&format!("{}{}", self.url, path))
.set("authorization", &format!("Bearer {}", self.token))
}
pub fn health(&self) -> Result<(), String> {
ureq::get(&format!("{}/health", self.url))
.timeout(std::time::Duration::from_secs(15))
.call()
.map(|_| ())
.map_err(|e| format!("cannot reach {}: {}", self.url, describe(e)))
}
pub fn check(&self) -> Result<RemoteCounts, String> {
self.health()?;
self.counts()
.map_err(|e| format!("reached {} but {}", self.url, e))
}
pub fn counts(&self) -> Result<RemoteCounts, String> {
Ok(self.export()?.live_counts())
}
pub fn export(&self) -> Result<RemoteGraph, String> {
let resp = self
.get(&format!("/export?workspace={}", urlencode(&self.workspace)))
.timeout(std::time::Duration::from_secs(300))
.call()
.map_err(|e| self.describe(e))?;
resp.into_json::<RemoteGraph>()
.map_err(|e| format!("the server's response was not a graph: {e}"))
}
pub fn board_post(&self, body: &Value) -> Result<Value, String> {
let req = self.post("/messages");
let req = match &self.repo_roots {
Some(roots) => req.set("x-deciduous-repo-roots", &roots.join(",")),
None => req,
};
req.timeout(std::time::Duration::from_secs(30))
.send_json(body.clone())
.map_err(|e| self.board_error("POST", e))?
.into_json::<Value>()
.map_err(|e| format!("the server's answer to POST /messages was not JSON: {e}"))
}
pub fn board_read(&self, params: &[(&str, String)]) -> Result<Value, String> {
let query: Vec<String> = params
.iter()
.map(|(k, v)| format!("{k}={}", urlencode(v)))
.collect();
self.get(&format!("/messages?{}", query.join("&")))
.timeout(std::time::Duration::from_secs(30))
.call()
.map_err(|e| self.board_error("GET", e))?
.into_json::<Value>()
.map_err(|e| format!("the server's answer to GET /messages was not JSON: {e}"))
}
fn board_error(&self, method: &str, e: ureq::Error) -> String {
match e {
ureq::Error::Status(404, _) => format!(
"{method} {}/messages answered 404: this server has no message board. \
The board needs deciduous server {} or later; upgrade the server. \
Nothing was written locally instead, because the other agents on this \
project read the server.",
self.url,
crate::board::SERVER_VERSION_NEEDED
),
e => format!("{method} {}/messages: {}", self.url, describe(e)),
}
}
pub fn import(&self, mut graph: Value) -> Result<ImportReport, String> {
if let Some(why) = self.write_blocker() {
return Err(format!("nothing was sent: {why}"));
}
wire_dates(&mut graph);
let payload = serde_json::json!({
"workspace": self.workspace,
"repo_roots": self.repo_roots,
"graph": graph,
});
self.post("/import")
.timeout(std::time::Duration::from_secs(600))
.send_json(payload)
.map_err(|e| self.describe(e))?
.into_json::<ImportReport>()
.map_err(|e| format!("the server's response was not an import report: {e}"))
}
}
pub fn missing_on_server(local: &Value, server: &RemoteGraph) -> (Value, usize, usize) {
use std::collections::{HashMap, HashSet};
let have_nodes: HashSet<&str> = server.nodes.iter().map(|n| n.change_id.as_str()).collect();
let dead = server.tombstones();
let ends = server.edge_ends();
let have_edges: HashSet<(&str, &str, &str)> = ends
.iter()
.map(|e| (e.from, e.to, e.edge.edge_type.as_str()))
.collect();
let have_docs: HashSet<&str> = server
.documents
.iter()
.filter_map(|d| d["change_id"].as_str())
.collect();
let empty = Vec::new();
let nodes = local["nodes"].as_array().unwrap_or(&empty);
let by_id: HashMap<i64, &str> = nodes
.iter()
.filter_map(|n| Some((n["id"].as_i64()?, n["change_id"].as_str()?)))
.collect();
let send_nodes: Vec<Value> = nodes
.iter()
.filter(|n| {
n["change_id"]
.as_str()
.is_some_and(|c| !have_nodes.contains(c))
})
.cloned()
.collect();
let mut send_edges = Vec::new();
for e in local["edges"].as_array().unwrap_or(&empty) {
let end = |id: &str, cid: &str| {
e[id]
.as_i64()
.and_then(|i| by_id.get(&i).copied())
.or_else(|| e[cid].as_str())
};
let (Some(f), Some(t)) = (
end("from_node_id", "from_change_id"),
end("to_node_id", "to_change_id"),
) else {
continue;
};
let kind = e["edge_type"].as_str().unwrap_or("leads_to");
if have_edges.contains(&(f, t, kind)) || dead.contains_key(f) || dead.contains_key(t) {
continue;
}
let mut e = e.clone();
e["from_change_id"] = Value::String(f.to_string());
e["to_change_id"] = Value::String(t.to_string());
send_edges.push(e);
}
let send_docs: Vec<Value> = local["documents"]
.as_array()
.unwrap_or(&empty)
.iter()
.filter(|d| {
d["change_id"]
.as_str()
.is_some_and(|c| !have_docs.contains(c))
})
.cloned()
.collect();
let (n, m) = (send_nodes.len(), send_edges.len());
(
serde_json::json!({"nodes": send_nodes, "edges": send_edges, "documents": send_docs}),
n,
m,
)
}
pub type Seeded = (Option<ImportReport>, Vec<DeletedOnServer>, Vec<String>);
pub fn push_missing(remote: &Remote, graph: &Value) -> Result<Seeded, String> {
let server = remote.export()?;
let deleted = deleted_on_server(graph, &server);
let (mut missing, _, _) = missing_on_server(graph, &server);
let withheld = withhold_nul(&mut missing);
let nodes = missing["nodes"].as_array().map_or(0, Vec::len);
let edges = missing["edges"].as_array().map_or(0, Vec::len);
let docs = missing["documents"].as_array().cloned().unwrap_or_default();
if nodes == 0 && edges == 0 && docs.is_empty() {
return Ok((None, deleted, withheld));
}
if !docs.is_empty() {
let dir = Database::db_path()
.parent()
.map(|p| p.join("documents"))
.ok_or("the database path has no directory, so there are no documents to read")?;
for d in &docs {
let (Some(hash), Some(file)) =
(d["content_hash"].as_str(), d["storage_filename"].as_str())
else {
return Err(format!(
"a local document has no content_hash or storage_filename: {d}"
));
};
let path = dir.join(file);
let bytes = std::fs::read(&path).map_err(|e| {
format!(
"document {} ({}) cannot be sent: {e}. Nothing was sent.",
d["original_filename"].as_str().unwrap_or("?"),
path.display()
)
})?;
remote.upload_blob(hash, &bytes, d["mime_type"].as_str())?;
}
}
remote.import(missing).map(|r| (Some(r), deleted, withheld))
}
fn wire_dates(graph: &mut Value) {
for (kind, keys) in [
("nodes", &["created_at", "updated_at"][..]),
("edges", &["created_at"][..]),
] {
for row in graph[kind].as_array_mut().into_iter().flatten() {
for key in keys {
if let Some(s) = row[*key].as_str() {
let wired = records::wire_ts(s);
row[*key] = Value::String(wired);
}
}
}
}
}
pub fn row_nul(row: &Value) -> Option<String> {
fn find(v: &Value, path: &str) -> Option<String> {
match v {
Value::String(s) if s.contains('\0') => Some(path.to_string()),
Value::String(s) if path.ends_with("_json") => serde_json::from_str::<Value>(s)
.ok()
.and_then(|inner| find(&inner, path.trim_end_matches("_json"))),
Value::Object(m) => m.iter().find_map(|(k, x)| {
let at = if path.is_empty() {
k.clone()
} else {
format!("{path}.{k}")
};
if k.contains('\0') {
Some(format!("the key {at:?}"))
} else {
find(x, &at)
}
}),
Value::Array(a) => a
.iter()
.enumerate()
.find_map(|(i, x)| find(x, &format!("{path}.{i}"))),
_ => None,
}
}
find(row, "")
}
fn withhold_nul(missing: &mut Value) -> Vec<String> {
let mut out = Vec::new();
let mut gone = std::collections::HashSet::new();
if let Some(nodes) = missing["nodes"].as_array_mut() {
nodes.retain(|n| match row_nul(n) {
None => true,
Some(at) => {
let cid = n["change_id"].as_str().unwrap_or_default().to_string();
out.push(format!(
"{} {} \"{}\" (NUL at {at})",
n["node_type"].as_str().unwrap_or("node"),
cid.chars().take(8).collect::<String>(),
n["title"].as_str().unwrap_or_default().replace('\0', "\\0")
));
gone.insert(cid);
false
}
});
}
if let Some(edges) = missing["edges"].as_array_mut() {
edges.retain(|e| {
let end = |k: &str| e[k].as_str().is_some_and(|c| gone.contains(c));
let why = row_nul(e).map(|at| format!("NUL at {at}")).or_else(|| {
(end("from_change_id") || end("to_change_id"))
.then(|| "an endpoint is withheld".to_string())
});
match why {
None => true,
Some(why) => {
out.push(format!(
"edge {} -> {} ({why})",
e["from_change_id"]
.as_str()
.unwrap_or("?")
.chars()
.take(8)
.collect::<String>(),
e["to_change_id"]
.as_str()
.unwrap_or("?")
.chars()
.take(8)
.collect::<String>()
));
false
}
}
});
}
if let Some(docs) = missing["documents"].as_array_mut() {
docs.retain(|d| match row_nul(d) {
None => true,
Some(at) => {
out.push(format!(
"document {} (NUL at {at})",
d["original_filename"]
.as_str()
.unwrap_or("?")
.replace('\0', "\\0")
));
false
}
});
}
out
}
pub fn print_withheld(withheld: &[String]) {
use colored::Colorize;
if withheld.is_empty() {
return;
}
eprintln!(
"{} {} row(s) hold a NUL character, which the server cannot store; they were not sent \
(the other rows were):",
"Not sent:".red().bold(),
withheld.len()
);
for w in withheld {
eprintln!(" {w}");
}
eprintln!(
"Change that text here (a prompt: `deciduous prompt <id> ...`), then run \
`deciduous remote push --seed` again."
);
}
#[derive(Debug, Clone, PartialEq)]
pub struct DeletedOnServer {
pub id: i64,
pub change_id: String,
pub title: String,
pub deleted_at: String,
}
pub fn deleted_on_server(local: &Value, server: &RemoteGraph) -> Vec<DeletedOnServer> {
let dead = server.tombstones();
local["nodes"]
.as_array()
.map(|nodes| {
nodes
.iter()
.filter_map(|n| {
let cid = n["change_id"].as_str()?;
let at = dead.get(cid)?;
Some(DeletedOnServer {
id: n["id"].as_i64().unwrap_or_default(),
change_id: cid.to_string(),
title: n["title"].as_str().unwrap_or_default().to_string(),
deleted_at: at.to_string(),
})
})
.collect()
})
.unwrap_or_default()
}
pub fn counts_without(local: &Value, deleted: &[DeletedOnServer]) -> (usize, usize) {
use std::collections::HashSet;
let gone: HashSet<i64> = deleted.iter().map(|d| d.id).collect();
let nodes = local["nodes"].as_array().map_or(0, |n| {
n.iter()
.filter(|n| !n["id"].as_i64().is_some_and(|i| gone.contains(&i)))
.count()
});
let edges = local["edges"].as_array().map_or(0, |e| {
e.iter()
.filter(|e| {
!e["from_node_id"]
.as_i64()
.is_some_and(|i| gone.contains(&i))
&& !e["to_node_id"].as_i64().is_some_and(|i| gone.contains(&i))
})
.count()
});
(nodes, edges)
}
pub fn report_refused(r: &ImportReport, graph: &Value) {
use colored::Colorize;
if r.nodes.refused_deleted == 0 && r.edges.refused_deleted == 0 {
return;
}
eprintln!(
"{} the server refused {} node(s) and {} edge(s) because it has deleted them or \
their endpoints. `deciduous remote pull` applies those deletions here.",
"Note:".yellow(),
r.nodes.refused_deleted,
r.edges.refused_deleted
);
for ex in &r.nodes.refused_deleted_examples {
let cid = ex["change_id"].as_str().unwrap_or_default();
let local = graph["nodes"]
.as_array()
.and_then(|n| n.iter().find(|n| n["change_id"] == cid));
eprintln!(
" {} \"{}\" deleted on the server at {}",
local
.and_then(|n| n["id"].as_i64())
.map_or_else(|| cid.to_string(), |i| i.to_string()),
local.and_then(|n| n["title"].as_str()).unwrap_or("?"),
ex["deleted_at"].as_str().unwrap_or("?")
);
}
}
#[derive(Debug, Default)]
pub struct ReplayReport {
pub sent: usize,
pub applied: usize,
pub already: usize,
pub rejected: Vec<(crate::oplog::Op, String)>,
pub unreadable: Vec<crate::oplog::Unreadable>,
pub left: usize,
}
#[derive(Debug, Clone)]
pub enum ReplayError {
Log(String),
Unreachable(String),
NoAnswer(String),
Server(String),
Failed(String),
Config(String),
}
impl std::fmt::Display for ReplayError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ReplayError::Log(e)
| ReplayError::Unreachable(e)
| ReplayError::NoAnswer(e)
| ReplayError::Server(e)
| ReplayError::Failed(e)
| ReplayError::Config(e) => f.write_str(e),
}
}
}
impl From<ReplayError> for String {
fn from(e: ReplayError) -> String {
e.to_string()
}
}
pub const REPLAY_BATCH: usize = 200;
fn no_answer(t: &ureq::Transport) -> bool {
use std::error::Error;
if t.kind() != ureq::ErrorKind::Io {
return false;
}
let mut source = t.source();
while let Some(e) = source {
if let Some(io) = e.downcast_ref::<std::io::Error>() {
return matches!(
io.kind(),
std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock
);
}
source = e.source();
}
t.to_string().contains("timed out")
}
impl Remote {
fn describe(&self, e: ureq::Error) -> String {
match e {
ureq::Error::Status(409, _) => self.claim_refused(),
e => describe(e),
}
}
fn claim_refused(&self) -> String {
if self.repo_roots.as_ref().is_some_and(|r| r.is_empty()) {
return format!(
"workspace \"{ws}\" on {url} belongs to a repository, and this one has no commit yet, \
so the server cannot tell whether it is that repository or an unrelated project \
also called {ws}; nothing it sends is accepted until it can.\n\n\
Make the first commit (a clone of that repository has its commits already), or give \
this project its own workspace:\n\n \
deciduous remote init {url} --workspace <another-name>",
ws = self.workspace,
url = self.url
);
}
format!(
"workspace \"{ws}\" on {url} belongs to another repository: an unrelated project \
whose directory is also called {ws} (the name is the directory name, lowercased) \
claimed it first, and this repository shares none of its root commits.\n\n\
Give this project its own workspace:\n\n \
deciduous remote init {url} --workspace <another-name>\n\n\
or, if the two repositories really should write one graph, say so by naming it:\n\n \
deciduous remote init {url} --workspace {ws}",
ws = self.workspace,
url = self.url
)
}
pub fn write_blocker(&self) -> Option<String> {
if self.repo_roots.as_ref().is_some_and(|r| r.is_empty()) {
return Some(format!(
"this repository has no commit yet, so the server cannot tie workspace \"{ws}\" to \
it, and an unrelated repository also called {ws} could later claim the workspace \
with these writes in it. Nothing is sent until the first commit; the writes wait \
in the log, and the first command after that commit sends them.",
ws = self.workspace
));
}
None
}
pub fn unchecked_note(&self) -> Option<String> {
if self.shallow {
return Some(format!(
"this is a shallow clone: its root commit is not here, so the server cannot check \
that workspace \"{ws}\" belongs to this repository, and takes its writes \
unchecked, into whichever graph that name holds. `git fetch --unshallow` lets it \
check.",
ws = self.workspace
));
}
self.write_blocker()
}
pub fn claim(&self, adopt: bool) -> Result<String, String> {
#[derive(Deserialize)]
struct Reply {
claim: String,
}
let payload = serde_json::json!({
"workspace": self.workspace,
"repo_roots": self.repo_roots,
"adopt": adopt,
});
match self
.post("/claim")
.timeout(std::time::Duration::from_secs(30))
.send_json(payload)
{
Ok(resp) => resp
.into_json::<Reply>()
.map(|r| r.claim)
.map_err(|e| format!("the server's response was not a claim: {e}")),
Err(e) => Err(self.describe(e)),
}
}
pub fn upload_blob(&self, hash: &str, bytes: &[u8], mime: Option<&str>) -> Result<(), String> {
ureq::put(&format!("{}/blob/{hash}", self.url))
.set("authorization", &format!("Bearer {}", self.token))
.set("content-type", mime.unwrap_or("application/octet-stream"))
.timeout(std::time::Duration::from_secs(300))
.send_bytes(bytes)
.map(|_| ())
.map_err(|e| format!("uploading document {hash}: {}", self.describe(e)))
}
pub fn post_ops(
&self,
ops: &[crate::oplog::Op],
) -> Result<Vec<crate::oplog::Ack>, ReplayError> {
#[derive(Deserialize)]
struct Answer {
op_id: String,
result: String,
#[serde(default)]
reason: Option<String>,
}
#[derive(Deserialize)]
struct Reply {
results: Vec<Answer>,
}
let payload = serde_json::json!({
"workspace": self.workspace,
"repo_roots": self.repo_roots,
"ops": ops,
});
if self
.deadline
.is_some_and(|d| std::time::Instant::now() >= d)
{
return Err(ReplayError::Unreachable(format!(
"the replay after a write gets {} s in all, and they are spent; \
what was not answered waits",
QUICK_OPS_TIMEOUT.as_secs()
)));
}
let timeout = within(self.ops_timeout, self.deadline);
let reply: Reply = self
.post("/ops")
.timeout(timeout)
.send_json(payload)
.map_err(|e| match e {
ureq::Error::Transport(t) if no_answer(&t) => ReplayError::NoAnswer(format!(
"{} write(s) went to {} and no answer came within {:.1} s ({t}). The server \
may have applied some or all of them. They stay queued, and sending them \
again is safe: the server records the id of every op it applies and answers \
a resend \"duplicate\", so nothing is applied twice",
ops.len(),
self.url,
timeout.as_secs_f64()
)),
ureq::Error::Status(409, _) => ReplayError::Server(self.claim_refused()),
e @ ureq::Error::Status(500, _) => ReplayError::Failed(describe(e)),
e @ ureq::Error::Status(502..=504, _) => ReplayError::Unreachable(describe(e)),
e @ ureq::Error::Status(..) => ReplayError::Server(describe(e)),
e @ ureq::Error::Transport(_) => ReplayError::Unreachable(describe(e)),
})?
.into_json()
.map_err(|e| {
ReplayError::Server(format!("the server's response was not an ops report: {e}"))
})?;
let now = chrono::Utc::now().to_rfc3339();
let answers: std::collections::HashMap<String, Answer> = reply
.results
.into_iter()
.map(|a| (a.op_id.clone(), a))
.collect();
ops.iter()
.map(|op| {
let a = answers.get(&op.op_id).ok_or_else(|| {
ReplayError::Server(format!(
"the server answered for {} of {} ops and not for {} ({}); nothing was marked sent",
answers.len(),
ops.len(),
op.op_id,
op.body.describe()
))
})?;
match a.result.as_str() {
"applied" | "exists" | "absent" | "duplicate" | "rejected" => {}
other => {
return Err(ReplayError::Server(format!(
"the server answered {other:?} for {} ({}); this CLI knows applied, exists, absent, duplicate and rejected",
op.op_id,
op.body.describe()
)))
}
}
Ok(crate::oplog::Ack {
op_id: op.op_id.clone(),
result: a.result.clone(),
reason: a.reason.clone(),
at: now.clone(),
})
})
.collect()
}
}
pub use crate::oplog::{is_set_aside, SET_ASIDE};
#[derive(Default)]
struct Held {
keys: std::collections::HashMap<String, String>,
}
impl Held {
fn own(op: &crate::oplog::Op) -> Vec<String> {
use crate::oplog::OpBody::*;
match &op.body {
CreateNode { change_id, .. }
| UpdateNode { change_id, .. }
| DeleteNode { change_id, .. } => vec![format!("n:{change_id}")],
CreateEdge {
from_change_id,
to_change_id,
edge_type,
..
}
| DeleteEdge {
from_change_id,
to_change_id,
edge_type,
..
} => vec![format!("e:{from_change_id}|{to_change_id}|{edge_type}")],
AttachDocument { change_id, .. }
| DetachDocument { change_id, .. }
| DescribeDocument { change_id, .. } => vec![format!("d:{change_id}")],
}
}
fn needs(op: &crate::oplog::Op) -> Vec<String> {
let mut k = Self::own(op);
if let crate::oplog::OpBody::CreateEdge {
from_change_id,
to_change_id,
..
}
| crate::oplog::OpBody::DeleteEdge {
from_change_id,
to_change_id,
..
} = &op.body
{
k.push(format!("n:{from_change_id}"));
k.push(format!("n:{to_change_id}"));
}
if op.body.document().is_some() {
k.extend(op.body.change_ids().iter().map(|c| format!("n:{c}")));
}
k
}
fn add(&mut self, op: &crate::oplog::Op, by: &str) {
for k in Self::own(op) {
self.keys.entry(k).or_insert_with(|| by.to_string());
}
}
fn blocking(&self, op: &crate::oplog::Op) -> Option<String> {
Self::needs(op)
.iter()
.find_map(|k| self.keys.get(k).cloned())
}
fn hold_behind(&mut self, op: &crate::oplog::Op, by: &str) -> crate::oplog::Ack {
self.add(op, by);
crate::oplog::Ack {
op_id: op.op_id.clone(),
result: "rejected".into(),
reason: Some(format!(
"{SET_ASIDE}, not sent: it follows op {by}, which is set aside, and sent without \
it the server would refuse it for want of that op. `deciduous remote push` sends \
them again, in order; `deciduous remote push --drop {}` discards this one",
short_id(op)
)),
at: chrono::Utc::now().to_rfc3339(),
}
}
}
fn short_id(op: &crate::oplog::Op) -> String {
op.op_id.chars().take(8).collect()
}
pub fn replay(remote: &Remote, log: &crate::oplog::OpLog) -> Result<ReplayReport, ReplayError> {
replay_with(remote, log, true)
}
pub fn replay_with(
remote: &Remote,
log: &crate::oplog::OpLog,
progress: bool,
) -> Result<ReplayReport, ReplayError> {
let state = log.read().map_err(ReplayError::Log)?;
if !state.pending.is_empty() {
if let Some(why) = remote.write_blocker() {
return Err(ReplayError::Config(format!(
"{} write(s) not sent: {why}",
state.pending.len()
)));
}
}
let mut report = ReplayReport {
unreadable: state.unreadable.clone(),
..Default::default()
};
print_unreadable(&state, log);
let mut held = Held::default();
for (op, ack) in &state.rejected {
if ack.reason.as_deref().is_some_and(is_set_aside) {
held.add(op, &short_id(op));
}
}
let by_id: std::collections::HashMap<&str, &crate::oplog::Op> = state
.pending
.iter()
.map(|op| (op.op_id.as_str(), op))
.collect();
let tally = |report: &mut ReplayReport, acks: &[crate::oplog::Ack]| {
for ack in acks {
let Some(op) = by_id.get(ack.op_id.as_str()) else {
continue;
};
report.sent += 1;
match ack.result.as_str() {
"applied" => report.applied += 1,
"rejected" => report.rejected.push((
(*op).clone(),
ack.reason
.clone()
.unwrap_or_else(|| "no reason given".into()),
)),
_ => report.already += 1,
}
}
};
let mut sendable = Vec::with_capacity(state.pending.len());
let mut behind = Vec::new();
for op in &state.pending {
match held.blocking(op) {
Some(by) => behind.push(held.hold_behind(op, &by)),
None => sendable.push(op.clone()),
}
}
log.record_acks(&behind).map_err(ReplayError::Log)?;
tally(&mut report, &behind);
let total = sendable.len();
let show = progress && total > REPLAY_BATCH;
if show {
eprintln!(
"Sending {total} queued write(s) to {} in batches of {REPLAY_BATCH}",
remote.url
);
}
let mut slowest = std::time::Duration::ZERO;
let mut done = 0usize;
for batch in sendable.chunks(REPLAY_BATCH) {
if let Some(deadline) = remote.deadline {
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
if done > 0 && remaining < slowest * 3 / 2 {
report.left = total - done;
break;
}
}
upload_attached(remote, log, batch);
let started = std::time::Instant::now();
let mut sent = remote.post_ops(batch);
for _ in 0..2 {
let Err(ReplayError::NoAnswer(e)) = &sent else {
break;
};
if remote.deadline.is_some() {
break;
}
eprintln!(" {e}. Sending them again.");
sent = remote.post_ops(batch);
}
let (acks, stop) = match sent {
Ok(acks) => (acks, None),
Err(ReplayError::Failed(e)) => isolate(remote, batch, &e, &mut held),
Err(e) => return Err(e),
};
slowest = slowest.max(started.elapsed());
done += batch.len();
if show && stop.is_none() && done * 10 / total > (done - batch.len()) * 10 / total {
eprintln!(" {done} of {total} answered");
}
log.record_acks(&acks).map_err(ReplayError::Log)?;
tally(&mut report, &acks);
let delivered: Vec<&crate::oplog::Op> = acks
.iter()
.filter(|a| a.result == "applied" || a.result == "exists")
.filter_map(|a| by_id.get(a.op_id.as_str()).copied())
.collect();
if let Err(e) = log.record_delivered(&delivered) {
eprintln!("Warning: could not record what the server took: {e}");
}
if let Some(e) = stop {
log.compact().map_err(ReplayError::Log)?;
return Err(e);
}
}
log.compact().map_err(ReplayError::Log)?;
Ok(report)
}
fn upload_attached(remote: &Remote, log: &crate::oplog::OpLog, batch: &[crate::oplog::Op]) {
let Some(dir) = log.path().parent().map(|p| p.join("documents")) else {
return;
};
for op in batch {
let crate::oplog::OpBody::AttachDocument {
content_hash,
storage_filename,
mime_type,
original_filename,
..
} = &op.body
else {
continue;
};
let sent = std::fs::read(dir.join(storage_filename))
.map_err(|e| format!("{}: {e}", dir.join(storage_filename).display()))
.and_then(|bytes| remote.upload_blob(content_hash, &bytes, Some(mime_type)));
if let Err(e) = sent {
eprintln!(
"Warning: the bytes of \"{original_filename}\" did not reach the server ({e}); \
it is attached there without its content until they do (`deciduous remote push --seed` sends them)"
);
}
}
}
const MAX_SUSPECTS: usize = 3;
fn isolate(
remote: &Remote,
batch: &[crate::oplog::Op],
first: &str,
held: &mut Held,
) -> (Vec<crate::oplog::Ack>, Option<ReplayError>) {
let send = |op: &crate::oplog::Op| -> Result<crate::oplog::Ack, ReplayError> {
let once = remote.post_ops(std::slice::from_ref(op));
let twice = match once {
Err(ReplayError::Failed(_)) => remote.post_ops(std::slice::from_ref(op)),
other => other,
};
twice.map(|mut a| a.remove(0))
};
let mut acks = Vec::with_capacity(batch.len());
let mut suspects: Vec<(usize, String)> = Vec::new();
let mut deferred: Vec<usize> = Vec::new();
let mut queue: std::collections::VecDeque<usize> = (0..batch.len()).collect();
while let Some(i) = queue.pop_front() {
let op = &batch[i];
if let Some(by) = held.blocking(op) {
acks.push(held.hold_behind(op, &by));
continue;
}
let waits_on_suspect = suspects
.iter()
.map(|(s, _)| *s)
.chain(deferred.iter().copied())
.any(|s| {
let own = Held::own(&batch[s]);
Held::needs(op).iter().any(|k| own.contains(k))
});
if waits_on_suspect {
deferred.push(i);
continue;
}
match send(op) {
Ok(ack) => {
acks.push(ack);
for (s, _) in std::mem::take(&mut suspects) {
let op = &batch[s];
match send(op) {
Ok(ack) => acks.push(ack),
Err(ReplayError::Failed(e)) => {
held.add(op, &short_id(op));
acks.push(crate::oplog::Ack {
op_id: op.op_id.clone(),
result: "rejected".into(),
reason: Some(format!(
"{SET_ASIDE}, not answered by the server: the server failed \
on this op ({e}) each time it was sent alone, and answered \
the op sent between those tries, so this op, not the \
server, is what it fails on. It was taken out of the queue \
to let the writes after it through. `deciduous remote push` \
sends it again; `deciduous remote push --drop {}` discards it",
short_id(op)
)),
at: chrono::Utc::now().to_rfc3339(),
});
}
Err(e) => return (acks, Some(e)),
}
}
for d in std::mem::take(&mut deferred).into_iter().rev() {
queue.push_front(d);
}
}
Err(ReplayError::Failed(e)) => {
suspects.push((i, e));
if suspects.len() >= MAX_SUSPECTS {
return (
acks,
Some(ReplayError::Failed(format!(
"{first}; sent one at a time, the server failed on {} ops in a row, \
each twice, so the server is failing, not one op. Nothing was set \
aside; every write not answered waits",
suspects.len()
))),
);
}
}
Err(e) => return (acks, Some(e)),
}
}
if suspects.is_empty() {
return (acks, None);
}
(
acks,
Some(ReplayError::Failed(format!(
"{first}; sent alone, twice each, the server failed on {} and answered nothing \
after, so there is no telling whether they or the server are at fault. Nothing \
was set aside; they wait",
suspects
.iter()
.map(|(s, _)| batch[*s].body.describe())
.collect::<Vec<_>>()
.join(", ")
))),
)
}
pub fn print_unreadable(state: &crate::oplog::LogState, log: &crate::oplog::OpLog) {
use colored::Colorize;
if state.unreadable.is_empty() {
return;
}
eprintln!(
"{} {} line(s) of {} are not log entries and are not sent (the file is damaged; no server was involved):",
"Warning:".yellow(),
state.unreadable.len(),
log.path().display()
);
for u in &state.unreadable {
eprintln!(" {}", u.describe());
}
eprintln!(
"They are moved to {} the next time the log is compacted. If one is a write you want sent, \
repair its JSON and append it to the log again; every other line is sent as usual.",
log.unreadable_path().display()
);
}
pub fn settled_refusals(
rejected: &[(crate::oplog::Op, crate::oplog::Ack)],
nodes: &[crate::db::DecisionNode],
edges: &[crate::db::DecisionEdge],
server: &RemoteGraph,
) -> Vec<crate::oplog::Op> {
use crate::oplog::OpBody;
let here: std::collections::HashMap<&str, &crate::db::DecisionNode> =
nodes.iter().map(|n| (n.change_id.as_str(), n)).collect();
let there: std::collections::HashMap<&str, &RemoteNode> = server
.nodes
.iter()
.filter(|n| n.deleted_at.is_none())
.map(|n| (n.change_id.as_str(), n))
.collect();
let by_id: std::collections::HashMap<i32, &str> =
nodes.iter().map(|n| (n.id, n.change_id.as_str())).collect();
let edges_here: std::collections::HashSet<(String, String, String)> = edges
.iter()
.filter_map(|e| {
Some((
by_id.get(&e.from_node_id)?.to_string(),
by_id.get(&e.to_node_id)?.to_string(),
e.edge_type.clone(),
))
})
.collect();
let edges_there: std::collections::HashSet<(String, String, String)> = server
.edge_ends()
.iter()
.filter(|e| there.contains_key(e.from) && there.contains_key(e.to))
.map(|e| {
(
e.from.to_string(),
e.to.to_string(),
e.edge.edge_type.clone(),
)
})
.collect();
let same_node = |cid: &str| match (here.get(cid), there.get(cid)) {
(None, None) => true,
(Some(n), Some(s)) => node_fields_differ(n, s).is_empty(),
_ => false,
};
let same_edge = |f: &str, t: &str, k: &str| {
let key = (f.to_string(), t.to_string(), k.to_string());
edges_here.contains(&key) == edges_there.contains(&key)
};
rejected
.iter()
.filter(|(_, a)| !a.is_set_aside())
.filter(|(op, _)| match &op.body {
OpBody::UpdateNode {
change_id,
set,
metadata,
..
} => match (here.get(change_id.as_str()), there.get(change_id.as_str())) {
(None, None) => true,
(Some(n), Some(s)) => {
let differ = node_fields_differ(n, s);
set.keys()
.cloned()
.chain(metadata.keys().map(|k| format!("metadata.{k}")))
.all(|f| !differ.contains(&f))
}
_ => false,
},
OpBody::CreateNode { change_id, .. } | OpBody::DeleteNode { change_id, .. } => {
same_node(change_id)
}
OpBody::CreateEdge {
from_change_id,
to_change_id,
edge_type,
..
}
| OpBody::DeleteEdge {
from_change_id,
to_change_id,
edge_type,
..
} => same_edge(from_change_id, to_change_id, edge_type),
OpBody::AttachDocument { .. }
| OpBody::DetachDocument { .. }
| OpBody::DescribeDocument { .. } => false,
})
.map(|(op, _)| op.clone())
.collect()
}
fn node_fields_differ(n: &crate::db::DecisionNode, s: &RemoteNode) -> Vec<String> {
let mut out = Vec::new();
if n.title != s.title {
out.push("title".to_string());
}
if n.status != s.status {
out.push("status".to_string());
}
if n.description.as_deref().unwrap_or("") != s.description.as_deref().unwrap_or("") {
out.push("description".to_string());
}
if n.node_type != s.node_type {
out.push("type".to_string());
}
let here = metadata_map(n.metadata_json.as_deref());
let there = s
.metadata
.as_ref()
.and_then(|m| m.as_object().cloned())
.unwrap_or_default();
let keys: std::collections::BTreeSet<&String> = here.keys().chain(there.keys()).collect();
for k in keys {
if here.get(k) != there.get(k) {
out.push(format!("metadata.{k}"));
}
}
out
}
fn metadata_map(json: Option<&str>) -> serde_json::Map<String, Value> {
json.and_then(|m| serde_json::from_str::<Value>(m).ok())
.and_then(|v| v.as_object().cloned())
.unwrap_or_default()
}
pub fn drop_settled(
log: &crate::oplog::OpLog,
db: &Database,
server: &RemoteGraph,
) -> Result<Vec<crate::oplog::Op>, String> {
let state = log.read()?;
if state.rejected.iter().all(|(_, a)| a.is_set_aside()) {
return Ok(Vec::new());
}
let nodes = db
.get_all_nodes()
.map_err(|e| format!("reading the local graph: {e}"))?;
let edges = db
.get_all_edges()
.map_err(|e| format!("reading the local graph: {e}"))?;
let settled = settled_refusals(&state.rejected, &nodes, &edges, server);
let ids: std::collections::HashSet<String> = settled.iter().map(|o| o.op_id.clone()).collect();
log.drop_refused(&ids)?;
Ok(settled)
}
pub fn print_rejected(rejected: &[(crate::oplog::Op, String)], log: &crate::oplog::OpLog) {
use colored::Colorize;
let (aside, refused): (Vec<_>, Vec<_>) = rejected
.iter()
.partition(|(_, reason)| is_set_aside(reason));
let (behind, refused): (Vec<_>, Vec<_>) =
refused.into_iter().partition(|(op, _)| op.from_git());
if !behind.is_empty() {
eprintln!(
"{} {} edit(s) that came here through git are older than what the server holds, which keeps its own:",
"Behind the server:".yellow(),
behind.len()
);
for (op, reason) in behind.iter().take(10) {
eprintln!(" {} {}", op.body.describe(), reason.dimmed());
}
eprintln!(
"`deciduous remote pull` takes the server's values here and drops these; commit graph.json after it."
);
}
if !refused.is_empty() {
eprintln!(
"{} the server refused {} write(s):",
"Rejected:".red().bold(),
refused.len()
);
for (op, reason) in &refused {
eprintln!(" {} {}", op.body.describe(), reason.dimmed());
}
eprintln!(
"They stay in {} and the local graph keeps them. \
`deciduous remote status` lists them; `deciduous remote push --drop-rejected` discards them.",
log.path().display()
);
if refused
.iter()
.any(|(_, reason)| reason.contains("was deleted on the server"))
{
eprintln!(
"A node these writes touch was deleted on the server: `deciduous remote pull` removes it here and drops the refusals with it."
);
}
}
if !aside.is_empty() {
eprintln!(
"{} {} write(s) were not refused by the server; this machine held them back:",
"Set aside:".red().bold(),
aside.len()
);
for (op, reason) in &aside {
eprintln!(" {} {}", op.body.describe(), reason.dimmed());
}
eprintln!(
"They are writes the server has not got. They stay in {}; `deciduous remote push` \
sends them again (`--drop <op id>` discards one; `--drop-rejected` keeps them).",
log.path().display()
);
}
}
pub fn replay_after_write(log: &crate::oplog::OpLog) {
after_write(log, &mut None, true);
}
pub fn replay_after_write_quietly(log: &crate::oplog::OpLog, last: &mut Option<String>) {
after_write(log, last, false);
}
fn after_write(log: &crate::oplog::OpLog, last: &mut Option<String>, loud: bool) {
use colored::Colorize;
let data_dir = log.path().parent().unwrap_or(Path::new("."));
let remote = Remote::for_replay_after_write(data_dir);
let result = remote
.as_ref()
.map_err(|e| ReplayError::Config(e.clone()))
.and_then(|r| replay_with(r, log, loud));
let waiting = || match log.read() {
Ok(s) => format!("{} write(s)", s.pending.len()),
Err(e) => format!("an unknown number of writes (the log could not be read: {e})"),
};
let unsent = |text: String| crate::oplog::notice(crate::oplog::NoticeKind::Unsent, text);
match &result {
Ok(report) => {
if !report.rejected.is_empty() {
let aside = report
.rejected
.iter()
.filter(|(_, r)| is_set_aside(r))
.count();
let mut text = format!(
"{} earlier write(s) did not reach the shared server ({} refused by it, {} set \
aside by this machine because the server failed on them); they were made \
here and are not on the server:",
report.rejected.len(),
report.rejected.len() - aside,
aside
);
for (op, reason) in &report.rejected {
text.push_str(&format!("\n {}: {reason}", op.body.describe()));
}
text.push_str(&format!(
"\nThey stay in {}. `deciduous remote status` lists them.",
log.path().display()
));
unsent(text);
}
if !report.unreadable.is_empty() {
unsent(format!(
"{} line(s) of {} are not log entries and were not sent (the file is damaged): {}",
report.unreadable.len(),
log.path().display(),
report
.unreadable
.iter()
.map(|u| u.describe())
.collect::<Vec<_>>()
.join("; ")
));
}
}
Err(ReplayError::Unreachable(_) | ReplayError::NoAnswer(_)) => {}
Err(e) => unsent(format!(
"Writes made here did not reach the shared server, and will not until this is fixed: {e}\n{} wait in {}.",
waiting(),
log.path().display()
)),
}
const LEFT: &str = "some writes left for later";
let said_left = last.as_deref() == Some(LEFT);
let this = result.as_ref().err().map(|e| e.to_string());
let repeat = this.is_some() && *last == this;
*last = this;
if repeat {
return;
}
match result {
Ok(report) => {
print_rejected(&report.rejected, log);
if report.left > 0 && (loud || !said_left) {
eprintln!(
"{} {} queued write(s) reached the server in the {} s a write waits for it; \
the other {} wait in {}, unsent. `deciduous remote push` sends them now, \
with progress; each later write sends some.",
"Note:".yellow(),
report.sent,
QUICK_OPS_TIMEOUT.as_secs(),
report.left,
log.path().display(),
);
if !loud {
*last = Some(LEFT.into());
}
}
}
Err(ReplayError::NoAnswer(e)) => eprintln!(
"{} the local write succeeded; {e}.\n{} queued in {}. They are sent on the next \
write, or now with `deciduous remote push`.",
"Note:".yellow(),
waiting(),
log.path().display(),
),
Err(ReplayError::Unreachable(e)) => eprintln!(
"{} the local write succeeded but the server did not get it: {e}\n\
{} queued in {}. They are sent on the next write, or now with \
`deciduous remote push` once the server is reachable.",
"Warning:".yellow(),
waiting(),
log.path().display(),
),
Err(ReplayError::Server(e)) => eprintln!(
"{} the local write succeeded but the server refused it: {e}\n\
{} wait in {}, and every later write will be refused the same way \
until that is fixed. `deciduous remote status` lists them.",
"Warning:".yellow(),
waiting(),
log.path().display(),
),
Err(ReplayError::Failed(e)) => eprintln!(
"{} the local write succeeded but the server failed on the request: {e}\n\
{} wait in {}. They are sent again on the next write. An op is set aside only \
when the server fails on it alone, answers another, and fails on it again.",
"Warning:".yellow(),
waiting(),
log.path().display(),
),
Err(ReplayError::Config(e)) => eprintln!(
"{} the local write succeeded but could not be sent: {e}\n{} wait in {}.",
"Warning:".yellow(),
waiting(),
log.path().display(),
),
Err(ReplayError::Log(e)) => eprintln!(
"{} the local write succeeded but the log of writes for the server could not be \
read or written, so nothing was sent: {e}\n{} wait in {}.",
"Warning:".yellow(),
waiting(),
log.path().display(),
),
}
}
pub struct ReplayOnExit;
impl Drop for ReplayOnExit {
fn drop(&mut self) {
if let Some(log) = crate::oplog::take_appended() {
replay_after_write(&log);
}
crate::oplog::print_local_only();
}
}
#[derive(Debug, Default)]
pub struct ContentDiff {
pub local_nodes: usize,
pub server_nodes: usize,
pub local_edges: usize,
pub server_edges: usize,
pub only_local: Vec<(String, String)>,
pub deleted_on_server: Vec<(String, String)>,
pub only_server: Vec<(String, String)>,
pub differ: Vec<NodeDifference>,
pub edges_only_local: Vec<String>,
pub edges_only_server: Vec<String>,
pub server_edge_copies_drifted: usize,
}
impl ContentDiff {
pub fn is_empty(&self) -> bool {
self.only_local.is_empty()
&& self.deleted_on_server.is_empty()
&& self.only_server.is_empty()
&& self.differ.is_empty()
&& self.edges_only_local.is_empty()
&& self.edges_only_server.is_empty()
}
}
#[derive(Debug)]
pub struct NodeDifference {
pub change_id: String,
pub title: String,
pub here_newer: bool,
pub fields: Vec<(String, String, String)>,
}
pub fn content_diff(
nodes: &[crate::db::DecisionNode],
edges: &[crate::db::DecisionEdge],
server: &RemoteGraph,
) -> ContentDiff {
use std::collections::{BTreeMap, BTreeSet, HashMap};
let short = |c: &str| c.chars().take(8).collect::<String>();
let label = |t: &str, title: &str, cid: &str| format!("{t} {} \"{title}\"", short(cid));
let shown = |s: &str| {
let s: String = s.chars().take(60).collect();
format!("{s:?}")
};
let server_nodes: BTreeMap<&str, &RemoteNode> = server
.nodes
.iter()
.filter(|n| n.deleted_at.is_none())
.map(|n| (n.change_id.as_str(), n))
.collect();
let local_nodes: BTreeMap<&str, &crate::db::DecisionNode> =
nodes.iter().map(|n| (n.change_id.as_str(), n)).collect();
let mut d = ContentDiff {
local_nodes: local_nodes.len(),
server_nodes: server_nodes.len(),
..Default::default()
};
let tombstones: BTreeSet<&str> = server
.nodes
.iter()
.filter(|n| n.deleted_at.is_some())
.map(|n| n.change_id.as_str())
.collect();
for (cid, n) in &local_nodes {
let Some(s) = server_nodes.get(cid) else {
let row = (cid.to_string(), label(&n.node_type, &n.title, cid));
if tombstones.contains(cid) {
d.deleted_on_server.push(row);
} else {
d.only_local.push(row);
}
continue;
};
let mut fields = Vec::new();
let mut cmp = |name: &str, here: &str, there: &str| {
if here != there {
fields.push((name.to_string(), shown(here), shown(there)));
}
};
cmp("type", &n.node_type, &s.node_type);
cmp("title", &n.title, &s.title);
cmp("status", &n.status, &s.status);
cmp(
"description",
n.description.as_deref().unwrap_or(""),
s.description.as_deref().unwrap_or(""),
);
let local_meta: serde_json::Map<String, Value> = n
.metadata_json
.as_deref()
.and_then(|m| serde_json::from_str::<Value>(m).ok())
.and_then(|v| v.as_object().cloned())
.unwrap_or_default();
let server_meta = s
.metadata
.as_ref()
.and_then(|m| m.as_object().cloned())
.unwrap_or_default();
let keys: BTreeSet<&String> = local_meta.keys().chain(server_meta.keys()).collect();
for k in keys {
let here = local_meta.get(k).map(|v| v.to_string()).unwrap_or_default();
let there = server_meta
.get(k)
.map(|v| v.to_string())
.unwrap_or_default();
if here != there {
fields.push((format!("metadata.{k}"), here, there));
}
}
if !fields.is_empty() {
d.differ.push(NodeDifference {
change_id: cid.to_string(),
title: n.title.clone(),
here_newer: records::parse_ts(&n.updated_at) > records::parse_ts(&s.updated_at),
fields,
});
}
}
for (cid, s) in &server_nodes {
if !local_nodes.contains_key(cid) {
d.only_server
.push((cid.to_string(), label(&s.node_type, &s.title, cid)));
}
}
let by_id: HashMap<i32, &str> = nodes.iter().map(|n| (n.id, n.change_id.as_str())).collect();
let edge_label = |f: &str, t: &str, k: &str| format!("{} -> {} ({k})", short(f), short(t));
let local_edges: BTreeSet<(String, String, String)> = edges
.iter()
.filter_map(|e| {
let end = |id: i32, copy: &Option<String>| {
by_id
.get(&id)
.copied()
.or(copy.as_deref())
.map(str::to_string)
};
Some((
end(e.from_node_id, &e.from_change_id)?,
end(e.to_node_id, &e.to_change_id)?,
e.edge_type.clone(),
))
})
.collect();
let ends = server.edge_ends();
d.server_edge_copies_drifted = ends.iter().filter(|e| e.drifted).count();
let server_edges: BTreeSet<(String, String, String)> = ends
.iter()
.filter(|e| server_nodes.contains_key(e.from) && server_nodes.contains_key(e.to))
.map(|e| {
(
e.from.to_string(),
e.to.to_string(),
e.edge.edge_type.clone(),
)
})
.collect();
d.local_edges = local_edges.len();
d.server_edges = server_edges.len();
d.edges_only_local = local_edges
.difference(&server_edges)
.filter(|(f, t, _)| !tombstones.contains(f.as_str()) && !tombstones.contains(t.as_str()))
.map(|(f, t, k)| edge_label(f, t, k))
.collect();
d.edges_only_server = server_edges
.difference(&local_edges)
.map(|(f, t, k)| edge_label(f, t, k))
.collect();
d
}
pub fn documents_only_here(docs: &[crate::db::NodeDocument], server: &RemoteGraph) -> Vec<String> {
let have: std::collections::HashSet<&str> = server
.documents
.iter()
.filter_map(|d| d["change_id"].as_str())
.collect();
docs.iter()
.filter(|d| d.detached_at.is_none() && !have.contains(d.change_id.as_str()))
.map(|d| {
format!(
"{} on {}",
d.original_filename,
d.node_change_id.chars().take(8).collect::<String>()
)
})
.collect()
}
#[derive(Debug, Default)]
pub struct Repair {
pub ops: Vec<crate::oplog::OpBody>,
pub skipped: Vec<String>,
pub overwrites: Vec<String>,
pub deletes: usize,
}
pub fn repair_ops(
nodes: &[crate::db::DecisionNode],
server: &RemoteGraph,
base: &std::collections::HashMap<String, Value>,
overwrite_server: bool,
) -> Repair {
use serde_json::Map;
let live: std::collections::HashMap<&str, &RemoteNode> = server
.nodes
.iter()
.filter(|n| n.deleted_at.is_none())
.map(|n| (n.change_id.as_str(), n))
.collect();
let short = |c: &str| c.chars().take(8).collect::<String>();
let shown = |v: &Value| {
let s = match v {
Value::Null => "nothing".to_string(),
Value::String(s) => format!("{:?}", s.chars().take(60).collect::<String>()),
other => other.to_string(),
};
s.chars().take(80).collect::<String>()
};
let blank = |v: Value| match v {
Value::String(s) if s.is_empty() => Value::Null,
v => v,
};
let mut out = Repair::default();
for n in nodes {
let Some(s) = live.get(n.change_id.as_str()) else {
continue;
};
let b = base.get(&n.change_id);
let unchanged = |field: &str, now: &Value| -> Option<bool> {
let b = b?;
let then = match field.strip_prefix("metadata.") {
Some(k) => b["metadata"].get(k).cloned().unwrap_or(Value::Null),
None => b.get(field).cloned().unwrap_or(Value::Null),
};
Some(blank(then) == blank(now.clone()))
};
let mut set = Map::new();
let mut was = Map::new();
let mut meta = Map::new();
let mut was_meta = Map::new();
let mut consider = |field: String, here: Value, there: Value| {
if blank(here.clone()) == blank(there.clone()) {
return;
}
let safe = unchanged(&field, &there);
if safe != Some(true) {
out.overwrites.push(format!(
"{} \"{}\" {field}: the server has {}, this copy has {} ({})",
short(&n.change_id),
n.title,
shown(&there),
shown(&here),
if safe.is_none() {
"this copy has never pulled the node, so which side changed is unknown"
} else {
"changed on the server since this copy last pulled it"
}
));
if !overwrite_server {
return;
}
}
match field.strip_prefix("metadata.") {
Some(k) => {
meta.insert(k.to_string(), here);
was_meta.insert(k.to_string(), there);
}
None => {
set.insert(field.clone(), here);
was.insert(field, there);
}
}
};
let opt = |v: &Option<String>| v.clone().map(Value::String).unwrap_or(Value::Null);
consider(
"title".into(),
Value::String(n.title.clone()),
Value::String(s.title.clone()),
);
consider(
"status".into(),
Value::String(n.status.clone()),
Value::String(s.status.clone()),
);
consider(
"description".into(),
opt(&n.description),
opt(&s.description),
);
let here = metadata_map(n.metadata_json.as_deref());
let there = s
.metadata
.as_ref()
.and_then(|m| m.as_object().cloned())
.unwrap_or_default();
for (k, v) in &here {
if there.get(k) != Some(v) {
consider(
format!("metadata.{k}"),
v.clone(),
there.get(k).cloned().unwrap_or(Value::Null),
);
}
}
if n.node_type != s.node_type {
out.skipped.push(format!(
"{} type: here {}, server {} (a node's type cannot be changed over /ops)",
short(&n.change_id),
n.node_type,
s.node_type
));
}
for k in there.keys().filter(|k| !here.contains_key(*k)) {
out.skipped.push(format!(
"{} metadata.{k}: only on the server (an op merges keys and cannot remove one)",
short(&n.change_id)
));
}
if !set.is_empty() || !meta.is_empty() {
out.ops.push(crate::oplog::OpBody::UpdateNode {
change_id: n.change_id.clone(),
set,
metadata: meta,
was,
was_metadata: was_meta,
});
}
}
out
}
pub fn repair_deletes(
nodes: &[crate::db::DecisionNode],
store: &RecordStore,
server: &RemoteGraph,
log: &crate::oplog::LogState,
) -> Result<Vec<crate::oplog::OpBody>, String> {
let here: std::collections::HashSet<&str> =
nodes.iter().map(|n| n.change_id.as_str()).collect();
let logged: std::collections::HashSet<&str> = log
.pending
.iter()
.chain(log.rejected.iter().map(|(op, _)| op))
.filter_map(|op| match &op.body {
crate::oplog::OpBody::DeleteNode { change_id, .. } => Some(change_id.as_str()),
_ => None,
})
.collect();
let mut out = Vec::new();
for s in server.nodes.iter().filter(|n| n.deleted_at.is_none()) {
let cid = s.change_id.as_str();
if here.contains(cid) || logged.contains(cid) {
continue;
}
let Some(rec) = store.read_node(cid).map_err(|e| e.to_string())? else {
continue;
};
if !rec.is_tombstone() {
continue;
}
let mut was = serde_json::Map::new();
was.insert("title".into(), Value::String(rec.title.clone()));
was.insert(
"description".into(),
rec.description
.clone()
.map(Value::String)
.unwrap_or(Value::Null),
);
was.insert("status".into(), Value::String(rec.status.clone()));
out.push(crate::oplog::OpBody::DeleteNode {
change_id: rec.change_id.clone(),
node_type: Some(rec.node_type.clone()),
was,
was_metadata: rec
.metadata
.as_ref()
.and_then(|m| m.as_object().cloned())
.unwrap_or_default(),
});
}
Ok(out)
}
pub fn repair_detaches(
docs: &[crate::db::NodeDocument],
server: &RemoteGraph,
log: &crate::oplog::LogState,
) -> Vec<crate::oplog::OpBody> {
let there: std::collections::HashSet<&str> = server
.documents
.iter()
.filter_map(|d| d["change_id"].as_str())
.collect();
let logged: std::collections::HashSet<&str> = log
.pending
.iter()
.chain(log.rejected.iter().map(|(op, _)| op))
.filter(|op| matches!(op.body, crate::oplog::OpBody::DetachDocument { .. }))
.filter_map(|op| op.body.document())
.collect();
docs.iter()
.filter(|d| d.detached_at.is_some())
.filter(|d| there.contains(d.change_id.as_str()) && !logged.contains(d.change_id.as_str()))
.map(|d| crate::oplog::OpBody::DetachDocument {
change_id: d.change_id.clone(),
node_change_id: d.node_change_id.clone(),
})
.collect()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Moved {
Here,
Server,
Unknown,
}
pub fn moved(n: &crate::db::DecisionNode, s: &RemoteNode, base: Option<&Value>) -> Moved {
let Some(b) = base else {
return Moved::Unknown;
};
let blank = |v: Option<&Value>| match v {
None | Some(Value::Null) => Value::Null,
Some(Value::String(x)) if x.is_empty() => Value::Null,
Some(v) => v.clone(),
};
let there_meta = s.metadata.as_ref().and_then(|m| m.as_object());
for f in node_fields_differ(n, s) {
let (now, then) = match f.strip_prefix("metadata.") {
Some(k) => (
blank(there_meta.and_then(|m| m.get(k))),
blank(b["metadata"].get(k)),
),
None if f == "type" => continue,
None => {
let now = match f.as_str() {
"title" => Some(Value::String(s.title.clone())),
"status" => Some(Value::String(s.status.clone())),
_ => s.description.clone().map(Value::String),
};
(blank(now.as_ref()), blank(b.get(&f)))
}
};
if now != then {
return Moved::Server;
}
}
Moved::Here
}
pub fn known_server_state(
db: &Database,
log: Option<&crate::oplog::OpLog>,
) -> Result<std::collections::HashMap<String, Value>, String> {
let mut base = db
.remote_base()
.map_err(|e| format!("reading what was last pulled: {e}"))?;
let Some(log) = log else {
return Ok(base);
};
for op in log.delivered() {
match op.body {
crate::oplog::OpBody::CreateNode {
change_id,
title,
description,
status,
metadata,
..
} => {
base.insert(
change_id,
serde_json::json!({
"title": title, "description": description, "status": status,
"metadata": metadata,
}),
);
}
crate::oplog::OpBody::UpdateNode {
change_id,
set,
metadata,
..
} => {
let Some(b) = base.get_mut(&change_id).and_then(Value::as_object_mut) else {
continue;
};
for (k, v) in set {
b.insert(k, v);
}
let meta = b
.entry("metadata")
.or_insert_with(|| Value::Object(Default::default()));
if let Some(m) = meta.as_object_mut() {
for (k, v) in metadata {
m.insert(k, v);
}
}
}
_ => {}
}
}
Ok(base)
}
fn base_record(known: &Value) -> Value {
let mut out = serde_json::Map::new();
for key in ["title", "description", "status", "metadata"] {
match known.get(key) {
None | Some(Value::Null) => {}
Some(Value::Object(m)) if m.is_empty() => {}
Some(v) => {
out.insert(key.to_string(), v.clone());
}
}
}
Value::Object(out)
}
pub fn base_of(graph: &RemoteGraph) -> Vec<(String, Value)> {
graph
.nodes
.iter()
.filter(|n| n.deleted_at.is_none())
.map(|n| {
(
n.change_id.clone(),
serde_json::json!({
"title": n.title,
"description": n.description,
"status": n.status,
"metadata": n.metadata.clone().unwrap_or(Value::Object(Default::default())),
"updated_at": n.updated_at,
}),
)
})
.collect()
}
#[derive(Debug, Default)]
pub struct RemoteCounts {
pub nodes: usize,
pub edges: usize,
pub documents: usize,
}
#[derive(Debug, Deserialize)]
pub struct RemoteGraph {
#[serde(default)]
pub nodes: Vec<RemoteNode>,
#[serde(default)]
pub edges: Vec<RemoteEdge>,
#[serde(default)]
pub documents: Vec<Value>,
#[serde(default)]
pub edge_tombstones: Vec<RemoteEdgeTombstone>,
}
#[derive(Debug, Deserialize)]
pub struct RemoteEdgeTombstone {
pub from_change_id: String,
pub to_change_id: String,
pub edge_type: String,
pub deleted_at: String,
}
impl RemoteGraph {
pub fn tombstones(&self) -> std::collections::HashMap<&str, &str> {
self.nodes
.iter()
.filter_map(|n| Some((n.change_id.as_str(), n.deleted_at.as_deref()?)))
.collect()
}
pub fn edge_ends(&self) -> Vec<EdgeEnds<'_>> {
let key = |v: &Value| match v {
Value::String(s) => s.clone(),
v => v.to_string(),
};
let by_id: std::collections::HashMap<String, &str> = self
.nodes
.iter()
.filter_map(|n| Some((n.id.clone()?, n.change_id.as_str())))
.collect();
let node = |id: &Option<Value>| id.as_ref().and_then(|v| by_id.get(&key(v)).copied());
self.edges
.iter()
.filter_map(|e| {
let from = node(&e.from_node_id).or(e.from_change_id.as_deref())?;
let to = node(&e.to_node_id).or(e.to_change_id.as_deref())?;
let drifted = e.from_change_id.as_deref() != Some(from)
|| e.to_change_id.as_deref() != Some(to);
Some(EdgeEnds {
from,
to,
edge: e,
drifted,
})
})
.collect()
}
pub fn live_counts(&self) -> RemoteCounts {
RemoteCounts {
nodes: self.nodes.iter().filter(|n| n.deleted_at.is_none()).count(),
edges: self.edges.len(),
documents: self.documents.len(),
}
}
}
#[derive(Debug)]
pub struct EdgeEnds<'a> {
pub from: &'a str,
pub to: &'a str,
pub edge: &'a RemoteEdge,
pub drifted: bool,
}
#[derive(Debug, Deserialize)]
pub struct RemoteNode {
#[serde(default)]
pub id: Option<String>,
pub change_id: String,
pub node_type: String,
pub title: String,
pub description: Option<String>,
pub status: String,
#[serde(default)]
pub metadata: Option<Value>,
pub created_at: String,
pub updated_at: String,
#[serde(default)]
pub deleted_at: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct RemoteEdge {
#[serde(default)]
pub from_node_id: Option<Value>,
#[serde(default)]
pub to_node_id: Option<Value>,
pub from_change_id: Option<String>,
pub to_change_id: Option<String>,
pub edge_type: String,
#[serde(default)]
pub rationale: Option<String>,
pub created_at: String,
}
#[derive(Debug, Deserialize)]
pub struct ImportReport {
pub nodes: CountPair,
pub edges: EdgeReport,
#[serde(default)]
pub documents: Option<DocReport>,
}
#[derive(Debug, Deserialize)]
pub struct CountPair {
pub received: usize,
pub upserted: usize,
#[serde(default)]
pub refused_deleted: usize,
#[serde(default)]
pub refused_deleted_examples: Vec<Value>,
}
#[derive(Debug, Deserialize)]
pub struct EdgeReport {
pub received: usize,
pub upserted: usize,
pub unresolved: usize,
#[serde(default)]
pub stale_change_ids: usize,
#[serde(default)]
pub refused_deleted: usize,
}
#[derive(Debug, Deserialize)]
pub struct DocReport {
pub received: usize,
pub upserted: usize,
#[serde(default)]
pub content_missing: usize,
}
pub fn pull(remote: &Remote, db: &Database, store: &RecordStore) -> Result<PullReport, String> {
let graph = remote.export()?;
let mut written = 0usize;
let git = records::reconcile(db, store, false)?;
let log = db.oplog();
db.set_oplog(None);
let result = pull_server_rows(db, store, &graph, &mut written, log.as_ref());
db.set_oplog(log.clone());
let mut report = result?;
db.set_remote_base(&base_of(&graph))
.map_err(|e| format!("recording what was pulled: {e}"))?;
if let Some(log) = &log {
log.clear_delivered()?;
}
if let Some(log) = &log {
report.settled = drop_settled(log, db, &graph)?;
}
report.imported_nodes += git.nodes_imported;
report.updated_nodes += git.nodes_updated;
report.removed_nodes += git.nodes_deleted;
report.imported_edges += git.edges_imported;
report.removed_edges += git.edges_deleted;
Ok(report)
}
fn pull_server_rows(
db: &Database,
store: &RecordStore,
graph: &RemoteGraph,
written: &mut usize,
log: Option<&crate::oplog::OpLog>,
) -> Result<PullReport, String> {
let written_before = *written;
let refused: std::collections::HashMap<String, std::collections::BTreeSet<String>> = match log {
Some(log) => refused_fields(&log.read()?.rejected),
None => Default::default(),
};
let known = known_server_state(db, log)?;
let result = store.batch(|| -> Result<usize, String> {
for n in graph.nodes.iter().filter(|n| n.deleted_at.is_none()) {
let Some(fields) = refused.get(&n.change_id) else {
continue;
};
let Some(mut rec) = store.read_node(&n.change_id).map_err(|e| e.to_string())? else {
continue;
};
if rec.is_tombstone() {
continue;
}
let before = rec.clone();
for f in fields {
match f.as_str() {
"title" => rec.title = n.title.clone(),
"status" => rec.status = n.status.clone(),
"description" => rec.description = n.description.clone(),
key => {
let Some(key) = key.strip_prefix("metadata.") else {
continue;
};
let mut meta = rec
.metadata
.as_ref()
.and_then(|m| m.as_object().cloned())
.unwrap_or_default();
match n.metadata.as_ref().and_then(|m| m.get(key)) {
Some(v) => meta.insert(key.to_string(), v.clone()),
None => meta.remove(key),
};
rec.metadata = Some(Value::Object(meta));
}
}
}
if rec != before {
rec.updated_at = (records::parse_ts(&before.updated_at)
+ chrono::Duration::milliseconds(1))
.to_rfc3339();
store.write_node(&rec).map_err(|e| e.to_string())?;
*written += 1;
}
}
for n in &graph.nodes {
let rec = NodeRecord {
change_id: n.change_id.clone(),
node_type: n.node_type.clone(),
title: n.title.clone(),
description: n.description.clone(),
status: n.status.clone(),
metadata: n.metadata.clone(),
created_at: n.created_at.clone(),
updated_at: n.updated_at.clone(),
author: None,
deleted_at: n.deleted_at.clone(),
extra: Default::default(),
};
let base = n
.deleted_at
.is_none()
.then(|| known.get(&n.change_id).map(base_record))
.flatten();
if store
.absorb_node(&rec, base.as_ref())
.map_err(|e| e.to_string())?
{
*written += 1;
}
}
for EdgeEnds {
from, to, edge: e, ..
} in graph.edge_ends()
{
let rec = EdgeRecord {
edge_id: records::edge_id(from, to, &e.edge_type),
from_change_id: from.to_string(),
to_change_id: to.to_string(),
edge_type: e.edge_type.clone(),
rationale: e.rationale.clone(),
weight: None,
created_at: e.created_at.clone(),
author: None,
deleted_at: None,
extra: Default::default(),
};
if store.absorb_edge(&rec).map_err(|e| e.to_string())? {
*written += 1;
}
}
for t in &graph.edge_tombstones {
let id = records::edge_id(&t.from_change_id, &t.to_change_id, &t.edge_type);
let Some(mut rec) = store.read_edge(&id).map_err(|e| e.to_string())? else {
continue;
};
if rec.is_tombstone() {
continue;
}
rec.deleted_at = Some(t.deleted_at.clone());
if store.absorb_edge(&rec).map_err(|e| e.to_string())? {
*written += 1;
}
}
Ok(*written - written_before)
});
let records_written = match result {
Ok(Ok(n)) => n,
Ok(Err(e)) => return Err(e),
Err(e) => return Err(format!("record store: {e}")),
};
let report = records::reconcile(db, store, false)?;
let local = db
.get_graph()
.map_err(|e| format!("reading the local graph: {e}"))
.and_then(|g| {
serde_json::to_value(&g).map_err(|e| format!("serializing the local graph: {e}"))
})?;
let overridden = deleted_on_server(&local, graph);
overridden.iter().try_for_each(|d| {
db.delete_node(d.id as i32, false)
.map(|_| ())
.map_err(|e| format!("deleting node {} \"{}\": {e}", d.id, d.title))
})?;
for d in &overridden {
let scrubbed = store
.read_node(&d.change_id)
.map_err(|e| format!("record store: {e}"))?
.map(|mut rec| {
rec.title = String::new();
rec.description = None;
rec.metadata = None;
rec
});
if let Some(rec) = scrubbed {
store
.write_node(&rec)
.map_err(|e| format!("record store: {e}"))?;
}
}
let dead: std::collections::HashSet<String> = graph
.nodes
.iter()
.filter(|n| n.deleted_at.is_some())
.map(|n| n.change_id.clone())
.collect();
let dropped_rejected = match log {
Some(log) if !dead.is_empty() => log.drop_rejected_touching(&dead)?,
_ => 0,
};
Ok(PullReport {
fetched_nodes: graph
.nodes
.iter()
.filter(|n| n.deleted_at.is_none())
.count(),
fetched_tombstones: graph
.nodes
.iter()
.filter(|n| n.deleted_at.is_some())
.count(),
fetched_edges: graph.edges.len(),
records_written,
imported_nodes: report.nodes_imported,
updated_nodes: report.nodes_updated,
removed_nodes: report.nodes_deleted + overridden.len(),
imported_edges: report.edges_imported,
removed_edges: report.edges_deleted,
deleted_over_local_edits: overridden,
dropped_rejected,
settled: Vec::new(),
})
}
fn refused_fields(
rejected: &[(crate::oplog::Op, crate::oplog::Ack)],
) -> std::collections::HashMap<String, std::collections::BTreeSet<String>> {
let mut out: std::collections::HashMap<String, std::collections::BTreeSet<String>> =
Default::default();
for (op, ack) in rejected {
if ack.is_set_aside() {
continue;
}
if let crate::oplog::OpBody::UpdateNode {
change_id,
set,
metadata,
..
} = &op.body
{
let e = out.entry(change_id.clone()).or_default();
e.extend(set.keys().cloned());
e.extend(metadata.keys().map(|k| format!("metadata.{k}")));
}
}
out
}
#[derive(Debug, Clone)]
pub struct ServerIdMatch {
pub id: String,
pub change_id: String,
pub title: String,
pub deleted: bool,
}
pub fn find_by_server_id(remote: &Remote, reference: &str) -> Result<Vec<ServerIdMatch>, String> {
let r = reference.trim().to_ascii_lowercase();
Ok(remote
.export()?
.nodes
.into_iter()
.filter_map(|n| {
let id = n.id?;
id.to_ascii_lowercase()
.starts_with(&r)
.then(|| ServerIdMatch {
id,
change_id: n.change_id,
title: n.title,
deleted: n.deleted_at.is_some(),
})
})
.collect())
}
#[derive(Debug, Default)]
pub struct PullReport {
pub fetched_nodes: usize,
pub fetched_tombstones: usize,
pub fetched_edges: usize,
pub records_written: usize,
pub imported_nodes: usize,
pub updated_nodes: usize,
pub removed_nodes: usize,
pub imported_edges: usize,
pub removed_edges: usize,
pub deleted_over_local_edits: Vec<DeletedOnServer>,
pub dropped_rejected: usize,
pub settled: Vec<crate::oplog::Op>,
}
fn describe(e: ureq::Error) -> String {
match e {
ureq::Error::Status(code, resp) => {
let body = resp.into_string().unwrap_or_default();
let detail = serde_json::from_str::<Value>(&body)
.ok()
.and_then(|v| v.get("error").and_then(|e| e.as_str()).map(str::to_string))
.unwrap_or_else(|| body.chars().take(200).collect());
match code {
401 => format!("the server rejected the token (401). Check {TOKEN_ENV}."),
404 => "no such endpoint (404). Is the URL right, including any path prefix?"
.to_string(),
_ if detail.is_empty() => format!("the server returned {code}"),
_ => format!("the server returned {code}: {detail}"),
}
}
ureq::Error::Transport(t) => format!("transport error: {t}"),
}
}
fn urlencode(s: &str) -> String {
s.bytes()
.map(|b| match b {
b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
(b as char).to_string()
}
b => format!("%{b:02X}"),
})
.collect()
}
pub fn find_projects(root: &Path) -> Vec<std::path::PathBuf> {
fn walk(dir: &Path, depth: usize, out: &mut Vec<std::path::PathBuf>) {
if depth > 3 {
return;
}
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for entry in entries.flatten() {
let path = entry.path();
if !path.is_dir() {
continue;
}
let name = entry.file_name();
let name = name.to_string_lossy();
if name == ".deciduous" {
if path.join("deciduous.db").exists() || path.join("config.toml").exists() {
if let Some(project) = path.parent() {
out.push(project.to_path_buf());
}
}
continue;
}
if name == "node_modules" || name == "target" || name == "_build" || name == "deps" {
continue;
}
walk(&path, depth + 1, out);
}
}
let mut out = Vec::new();
walk(root, 0, &mut out);
out.sort();
out.dedup();
out
}
pub fn read_remote_url(project: &Path) -> Option<String> {
let text = std::fs::read_to_string(project.join(".deciduous").join("config.toml")).ok()?;
let doc: toml::Value = toml::from_str(&text).ok()?;
doc.get("remote")?.get("url")?.as_str().map(str::to_string)
}
pub fn write_remote_url(project: &Path, url: &str) -> Result<(), String> {
let dir = project.join(".deciduous");
let path = dir.join("config.toml");
let existing = std::fs::read_to_string(&path).unwrap_or_default();
let mut doc = crate::config::parse_with_table(&existing, "remote")
.map_err(|e| format!("{} is not valid TOML: {e}", path.display()))?;
let remote = doc["remote"].or_insert(toml_edit::table());
remote["url"] = toml_edit::value(url.to_string());
if remote.get("workspace").is_none() {
remote["workspace"] = toml_edit::value(workspace_for(project));
}
std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
std::fs::write(&path, doc.to_string()).map_err(|e| format!("{}: {e}", path.display()))
}
#[cfg(test)]
mod tests {
fn server(nodes: &[&str], edges: &[(&str, &str, &str)]) -> RemoteGraph {
RemoteGraph {
nodes: nodes
.iter()
.map(|c| RemoteNode {
id: None,
change_id: c.to_string(),
node_type: "goal".into(),
title: "t".into(),
description: None,
status: "pending".into(),
metadata: None,
created_at: "x".into(),
updated_at: "x".into(),
deleted_at: None,
})
.collect(),
edges: edges
.iter()
.map(|(f, t, k)| RemoteEdge {
from_node_id: None,
to_node_id: None,
from_change_id: Some(f.to_string()),
to_change_id: Some(t.to_string()),
edge_type: k.to_string(),
rationale: None,
created_at: "x".into(),
})
.collect(),
documents: vec![],
edge_tombstones: vec![],
}
}
#[test]
fn push_sends_only_what_the_server_lacks_and_never_an_existing_node() {
let local = serde_json::json!({
"nodes": [{"id": 1, "change_id": "a"}, {"id": 2, "change_id": "b"}, {"id": 3, "change_id": "c"}],
"edges": [
{"from_node_id": 1, "to_node_id": 2, "edge_type": "leads_to"},
{"from_node_id": 2, "to_node_id": 3, "edge_type": "leads_to"}
],
"documents": []
});
let (g, n, m) = missing_on_server(&local, &server(&["a", "b"], &[("a", "b", "leads_to")]));
assert_eq!((n, m), (1, 1));
assert_eq!(g["nodes"][0]["change_id"], "c");
assert_eq!(g["edges"][0]["from_change_id"], "b");
assert_eq!(g["edges"][0]["to_change_id"], "c");
}
#[test]
fn a_stale_stored_change_id_does_not_make_an_existing_edge_look_missing() {
let local = serde_json::json!({
"nodes": [{"id": 1, "change_id": "a"}, {"id": 2, "change_id": "b"}],
"edges": [{"from_node_id": 1, "to_node_id": 2, "from_change_id": "old-a", "to_change_id": "b", "edge_type": "leads_to"}]
});
let (_, n, m) = missing_on_server(&local, &server(&["a", "b"], &[("a", "b", "leads_to")]));
assert_eq!((n, m), (0, 0));
}
#[test]
fn status_and_seed_key_edges_by_the_nodes_the_rows_join() {
let mut g = server(&["a", "b"], &[("old-a", "b", "leads_to")]);
for n in &mut g.nodes {
n.id = Some(format!("srv-{}", n.change_id));
}
g.edges[0].from_node_id = Some(Value::String("srv-a".into()));
g.edges[0].to_node_id = Some(Value::String("srv-b".into()));
let local = serde_json::json!({
"nodes": [{"id": 1, "change_id": "a"}, {"id": 2, "change_id": "b"}],
"edges": [{"from_node_id": 1, "to_node_id": 2, "from_change_id": "old-a", "to_change_id": "b", "edge_type": "leads_to"}]
});
assert_eq!(missing_on_server(&local, &g).2, 0, "seed sends nothing");
let node = |id, cid: &str| crate::db::DecisionNode {
id,
change_id: cid.into(),
node_type: "goal".into(),
title: "t".into(),
description: None,
status: "pending".into(),
created_at: "x".into(),
updated_at: "x".into(),
metadata_json: None,
};
let edge = crate::db::DecisionEdge {
id: 1,
from_node_id: 1,
to_node_id: 2,
from_change_id: Some("old-a".into()),
to_change_id: Some("b".into()),
edge_type: "leads_to".into(),
weight: None,
rationale: None,
created_at: "x".into(),
};
let d = content_diff(&[node(1, "a"), node(2, "b")], &[edge], &g);
assert!(d.edges_only_local.is_empty(), "{:?}", d.edges_only_local);
assert!(d.edges_only_server.is_empty(), "{:?}", d.edges_only_server);
assert_eq!(d.server_edge_copies_drifted, 1);
assert!(d.is_empty(), "a drifted copy is not a difference");
}
use super::*;
#[test]
fn workspace_falls_back_outside_a_git_repo() {
let dir = std::env::temp_dir().join("deciduous-remote-not-a-repo");
std::fs::create_dir_all(&dir).unwrap();
assert_eq!(workspace_for(&dir), FALLBACK_WORKSPACE);
}
#[test]
fn workspace_names_are_lowercased() {
let name = "MixedCase";
assert_eq!(name.to_lowercase(), "mixedcase");
}
#[test]
fn the_global_token_is_escaped_so_it_cannot_widen_a_query() {
assert_eq!(urlencode("*"), "%2A");
assert_eq!(urlencode("blog"), "blog");
assert_eq!(urlencode("a b"), "a%20b");
}
#[test]
fn non_ascii_is_encoded_as_utf8_bytes() {
assert_eq!(urlencode("café"), "caf%C3%A9");
assert_eq!(urlencode("ü"), "%C3%BC");
assert_eq!(urlencode("日"), "%E6%97%A5");
}
#[test]
fn adopt_skips_vendored_trees() {
let root = std::env::temp_dir().join(format!("deciduous-adopt-{}", std::process::id()));
let real = root.join("my-project").join(".deciduous");
let vendored = root
.join("my-project")
.join("node_modules")
.join("dep")
.join(".deciduous");
std::fs::create_dir_all(&real).unwrap();
std::fs::create_dir_all(&vendored).unwrap();
std::fs::write(real.join("config.toml"), "").unwrap();
std::fs::write(vendored.join("config.toml"), "").unwrap();
let found = find_projects(&root);
assert_eq!(found.len(), 1, "found: {found:?}");
assert!(found[0].ends_with("my-project"));
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn writing_a_remote_url_preserves_the_rest_of_the_file() {
let root = std::env::temp_dir().join(format!("deciduous-wr-{}", std::process::id()));
let dir = root.join(".deciduous");
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(
dir.join("config.toml"),
"# keep me\n[branch]\nmain_branches = [\"main\"]\n",
)
.unwrap();
write_remote_url(&root, "https://example.com/mcp").unwrap();
let after = std::fs::read_to_string(dir.join("config.toml")).unwrap();
assert!(after.contains("# keep me"), "comment lost:\n{after}");
assert!(after.contains("main_branches = [\"main\"]"));
assert_eq!(
read_remote_url(&root).as_deref(),
Some("https://example.com/mcp")
);
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_remote_without_a_url_explains_how_to_configure_one() {
let cfg = Config::default();
let err = Remote::resolve(&cfg, Path::new(".")).unwrap_err();
assert!(err.contains("deciduous remote init"), "got: {err}");
}
#[test]
fn events_url_converts_https_to_wss() {
let r = Remote {
url: "https://example.com/deciduous-mcp".to_string(),
workspace: "blog".to_string(),
token: "abc123".to_string(),
repo_roots: None,
shallow: false,
ops_timeout: OPS_TIMEOUT,
deadline: None,
};
assert_eq!(
r.events_url(),
"wss://example.com/deciduous-mcp/events?workspace=blog&token=abc123"
);
}
#[test]
fn events_url_converts_plain_http_to_ws_for_a_dev_server() {
let r = Remote {
url: "http://localhost:4111".to_string(),
workspace: "blog".to_string(),
token: "abc123".to_string(),
repo_roots: None,
shallow: false,
ops_timeout: OPS_TIMEOUT,
deadline: None,
};
assert_eq!(
r.events_url(),
"ws://localhost:4111/events?workspace=blog&token=abc123"
);
}
#[test]
fn events_url_encodes_a_workspace_name_with_special_characters() {
let r = Remote {
url: "https://example.com".to_string(),
workspace: "a b".to_string(),
token: "tok".to_string(),
repo_roots: None,
shallow: false,
ops_timeout: OPS_TIMEOUT,
deadline: None,
};
assert!(
r.events_url().contains("workspace=a%20b"),
"got: {}",
r.events_url()
);
}
#[test]
fn a_server_tombstone_is_not_counted_and_not_pushed_back() {
let mut graph = server(&["a", "b"], &[]);
graph.nodes[1].deleted_at = Some("2026-09-23T17:38:04Z".into());
let c = graph.live_counts();
assert_eq!((c.nodes, c.edges), (1, 0));
let local = serde_json::json!({
"nodes": [{"id": 1, "change_id": "a"}, {"id": 2, "change_id": "b"}],
"edges": [],
"documents": []
});
let (_, n, _) = missing_on_server(&local, &graph);
assert_eq!(n, 0);
}
}