use crate::config;
use crate::repository::*;
use crate::Error;
use libanu::pristine::{Base32, ChannelRef, Hash, Merkle, MutTxnT, RemoteRef};
use libanu::DOT_DIR;
use libanu::{MutTxnTExt, TxnTExt};
use std::io::Write;
use std::path::{Path, PathBuf};
pub mod ssh;
use ssh::*;
pub mod local;
use local::*;
pub enum RemoteRepo {
Local(Local),
Ssh(Ssh),
Http(Http),
None,
}
pub struct Http {
pub url: String,
pub channel: String,
pub client: reqwest::Client,
pub name: String,
}
impl Repository {
pub async fn remote(&self, name: &str, channel: &str, no_cert_check: bool) -> Result<RemoteRepo, anyhow::Error> {
match self.config.remotes.get(name) {
Some(config::Remote::Local { ref local }) => {
let mut dot_dir = Path::new(local).join(DOT_DIR);
let changes_dir = dot_dir.join(CHANGES_DIR);
dot_dir.push(PRISTINE_DIR);
let pristine = libanu::pristine::sanakirja::Pristine::new(&dot_dir)?;
Ok(RemoteRepo::Local(Local {
channel: channel.to_string(),
changes_dir,
pristine,
root: Path::new(local).to_path_buf(),
name: name.to_string(),
}))
}
Some(config::Remote::Ssh(ref r)) => {
if let Some(ssh) = ssh_remote(&r.addr) {
Ok(RemoteRepo::Ssh(ssh.connect(name, channel).await?))
} else {
Err((Error::IncorrectRemote {
name: name.to_string(),
})
.into())
}
}
Some(config::Remote::Http { ref url }) => Ok(RemoteRepo::Http(Http {
url: url.to_string(),
channel: channel.to_string(),
client: reqwest::ClientBuilder::new()
.danger_accept_invalid_certs(no_cert_check)
.build()?,
name: name.to_string(),
})),
Some(config::Remote::None) => return Err(Error::UnknownRemoteType.into()),
None => unknown_remote(name, channel, no_cert_check).await,
}
}
}
pub async fn unknown_remote(name: &str, channel: &str, no_cert_check: bool) -> Result<RemoteRepo, anyhow::Error> {
if name.starts_with("http://") || name.starts_with("https://") {
debug!("unknown_remote, http = {:?}", name);
Ok(RemoteRepo::Http(Http {
url: name.to_string(),
channel: channel.to_string(),
client: reqwest::ClientBuilder::new()
.danger_accept_invalid_certs(no_cert_check)
.build()?,
name: name.to_string(),
}))
} else if let Some(ssh) = ssh_remote(name) {
debug!("unknown_remote, ssh = {:?}", ssh);
Ok(RemoteRepo::Ssh(ssh.connect(name, channel).await?))
} else {
let mut dot_dir = Path::new(name).join(DOT_DIR);
let changes_dir = dot_dir.join(CHANGES_DIR);
dot_dir.push(PRISTINE_DIR);
debug!("dot_dir = {:?}", dot_dir);
let pristine = libanu::pristine::sanakirja::Pristine::new(&dot_dir)?;
debug!("pristine done");
Ok(RemoteRepo::Local(Local {
channel: channel.to_string(),
changes_dir,
pristine,
root: Path::new(name).to_path_buf(),
name: name.to_string(),
}))
}
}
impl RemoteRepo {
fn name(&self) -> &str {
match *self {
RemoteRepo::Ssh(ref s) => s.name.as_str(),
RemoteRepo::Local(ref l) => l.name.as_str(),
RemoteRepo::Http(ref h) => h.name.as_str(),
RemoteRepo::None => unreachable!(),
}
}
pub async fn finish(&mut self) -> Result<(), anyhow::Error> {
match *self {
RemoteRepo::Ssh(ref mut s) => s.finish().await?,
_ => {}
}
Ok(())
}
pub async fn update_changelist<T: MutTxnT>(
&mut self,
txn: &mut T,
path: &[String],
) -> Result<RemoteRef<T>, anyhow::Error> {
debug!("update_changelist");
let name = self.name();
let mut remote = txn.open_or_create_remote(name).unwrap();
let n = self
.dichotomy_changelist(txn, &remote.borrow().remote)
.await?;
debug!("update changelist {:?}", n);
let v: Vec<_> = txn
.iter_remote(&remote.borrow().remote, n)
.filter_map(|(k, _)| if k >= n { Some(k) } else { None })
.collect();
for k in v {
debug!("deleting {:?}", k);
txn.del_remote(&mut remote, k)?;
}
self.download_changelist(txn, &mut remote, n, path).await?;
Ok(remote)
}
async fn dichotomy_changelist<T: MutTxnT>(
&mut self,
txn: &T,
remote: &T::Remote,
) -> Result<u64, anyhow::Error> {
let mut a = 0;
let (mut b, (_, state)) = if let Some(last) = txn.last_remote(remote) {
last
} else {
debug!("the local copy of the remote has no changes");
return Ok(0);
};
if let Some((_, s)) = self.get_state(Some(b)).await? {
if s == state {
// The local list is already up to date.
return Ok(b + 1);
}
}
// Else, find the last state we have in common with the
// remote, it might be older than the last known state (if
// changes were unrecorded on the remote).
while a < b {
let mid = (a + b) / 2;
let (mid, (_, state)) = txn.get_remote_state(remote, mid).unwrap();
let remote_state = self.get_state(Some(mid)).await?;
debug!("dichotomy {:?} {:?} {:?}", mid, state, remote_state);
if let Some((_, remote_state)) = remote_state {
if remote_state == state {
if a == mid {
return Ok(a + 1);
} else {
a = mid;
continue;
}
}
}
if b == mid {
break;
} else {
b = mid
}
}
Ok(a)
}
async fn get_state(
&mut self,
mid: Option<u64>,
) -> Result<Option<(u64, Merkle)>, anyhow::Error> {
match *self {
RemoteRepo::Local(ref mut l) => l.get_state(mid),
RemoteRepo::Ssh(ref mut s) => s.get_state(mid).await,
RemoteRepo::Http(ref h) => {
debug!("get_state {:?}", h.url);
let url = format!("{}/{}", h.url, DOT_DIR);
let q = if let Some(mid) = mid {
[("state", format!("{}", mid)),
("channel", h.channel.clone())]
} else {
[("state", String::new()),
("channel", h.channel.clone())]
};
let res = h.client.get(&url).query(&q).send().await?;
if res.status().is_success() {
let resp = res.bytes().await?;
let resp = std::str::from_utf8(&resp)?;
debug!("resp = {:?}", resp);
let mut s = resp.split(' ');
if let (Some(n), Some(m)) = (s.next().and_then(|s| s.parse().ok()), s.next().and_then(|m| Merkle::from_base32(m.as_bytes()))) {
Ok(Some((n, m)))
} else {
Ok(None)
}
} else {
Err((crate::Error::Remote {
msg: std::str::from_utf8(&res.bytes().await?)?.to_string(),
}).into())
}
}
RemoteRepo::None => unreachable!(),
}
}
pub async fn archive<W: std::io::Write>(
&mut self,
prefix: Option<String>,
state: Option<(Merkle, &[Hash])>,
mut w: W,
) -> Result<u64, anyhow::Error> {
match *self {
RemoteRepo::Local(ref mut l) => {
use libanu::pristine::TxnT;
debug!("archiving local repo");
let changes = libanu::changestore::filesystem::FileSystem::from_root(&l.root);
let mut tarball = libanu::output::Tarball::new(w, prefix);
let conflicts = if let Some((state, extra)) = state {
let mut txn = l.pristine.mut_txn_begin();
let mut channel = txn.load_channel(&l.channel).unwrap();
txn.archive_with_state(&changes, &mut channel, state, extra, &mut tarball)?
} else {
let txn = l.pristine.txn_begin()?;
let channel = txn.load_channel(&l.channel).unwrap();
txn.archive(&changes, &channel, &mut tarball)?
};
Ok(conflicts.len() as u64)
}
RemoteRepo::Ssh(ref mut s) => s.archive(prefix, state, w).await,
RemoteRepo::Http(ref h) => {
let url = h.url.clone() + "/" + DOT_DIR;
let res = h.client.get(&url).query(&[("channel", &h.channel)]);
let res = if let Some((ref state, ref extra)) = state {
let mut q = vec![("archive".to_string(), state.to_base32())];
if let Some(pre) = prefix {
q.push(("outputPrefix".to_string(), pre));
}
for e in extra.iter() {
q.push(("change".to_string(), e.to_base32()))
}
res.query(&q)
} else {
res
};
let res = res.send().await?;
use futures_util::StreamExt;
let mut stream = res.bytes_stream();
let mut conflicts = 0;
let mut n = 0;
while let Some(item) = stream.next().await {
let item = item?;
let mut off = 0;
while n < 8 && off < item.len() {
conflicts = (conflicts << 8) | (item[off] as u64);
off += 1;
n += 1
}
w.write_all(&item[off..])?;
}
Ok(conflicts as u64)
}
RemoteRepo::None => unreachable!(),
}
}
async fn download_changelist<T: MutTxnT>(
&mut self,
txn: &mut T,
remote: &mut RemoteRef<T>,
from: u64,
paths: &[String],
) -> Result<(), anyhow::Error> {
match *self {
RemoteRepo::Local(ref mut l) => l.download_changelist(txn, remote, from, paths),
RemoteRepo::Ssh(ref mut s) => s.download_changelist(txn, remote, from, paths).await,
RemoteRepo::Http(ref h) => {
let url = h.url.clone() + "/" + DOT_DIR;
let from_ = from.to_string();
let mut query = vec![("changelist", &from_), ("channel", &h.channel)];
for p in paths.iter() {
query.push(("path", p));
}
let res = h.client.get(&url).query(&query).send().await?;
let resp = res.bytes().await?;
if let Ok(data) = std::str::from_utf8(&resp) {
for l in data.lines() {
if !l.is_empty() {
let (n, h, m) = parse_line(l)?;
txn.put_remote(remote, n, (h, m))?;
} else {
break;
}
}
}
Ok(())
}
RemoteRepo::None => unreachable!(),
}
}
pub async fn upload_changes(
&mut self,
mut local: PathBuf,
to_channel: Option<&str>,
changes: &[Hash],
) -> Result<(), anyhow::Error> {
match self {
RemoteRepo::Local(ref mut l) => l.upload_changes(local, to_channel, changes),
RemoteRepo::Ssh(ref mut s) => s.upload_changes(local, to_channel, changes).await,
RemoteRepo::Http(ref h) => {
for c in changes {
libanu::changestore::filesystem::push_filename(&mut local, &c);
let url = h.url.clone() + "/" + DOT_DIR;
let change = std::fs::read(&local)?;
let to_channel = if let Some(ch) = to_channel {
format!("&to_channel={}", ch)
} else {
String::new()
};
h.client
.post(&url)
.query(&format!("apply={}{}", c.to_base32(), to_channel))
.body(change)
.send()
.await?;
libanu::changestore::filesystem::pop_filename(&mut local);
}
Ok(())
}
RemoteRepo::None => unreachable!(),
}
}
/// Start (and possibly complete) the download of a change.
pub async fn start_change_download(
&mut self,
c: libanu::pristine::Hash,
path: &mut PathBuf,
full: bool,
) -> Result<bool, anyhow::Error> {
debug!("start_change_download");
libanu::changestore::filesystem::push_filename(path, &c);
if std::fs::metadata(&path).is_ok() && !full {
debug!("metadata {:?} ok", path);
libanu::changestore::filesystem::pop_filename(path);
return Ok(false);
}
std::fs::create_dir_all(&path.parent().unwrap())?;
match *self {
RemoteRepo::Local(ref mut l) => l.start_change_download(c, path).await?,
RemoteRepo::Ssh(ref mut s) => s.start_change_download(c, full).await?,
RemoteRepo::Http(ref h) => {
let mut f = std::fs::File::create(&path)?;
let c32 = c.to_base32();
let url = format!("{}/{}", h.url, DOT_DIR);
let mut res = h.client.get(&url).query(&[("change", c32)]).send().await?;
while let Some(chunk) = res.chunk().await? {
f.write_all(&chunk)?;
}
}
RemoteRepo::None => unreachable!(),
}
libanu::changestore::filesystem::pop_filename(path);
Ok(true)
}
pub async fn wait_downloads(
&mut self,
changes_dir: &Path,
hashes: &[libanu::pristine::Hash],
send: &mut tokio::sync::mpsc::Sender<libanu::pristine::Hash>,
) -> Result<(), anyhow::Error> {
if hashes.is_empty() {
return Ok(());
}
if let RemoteRepo::Ssh(ref mut s) = *self {
s.wait_downloads(changes_dir, hashes, send).await?
} else {
for h in hashes {
send.send(*h).await?
}
}
Ok(())
}
pub async fn pull<T: MutTxnTExt + TxnTExt>(
&mut self,
repo: &mut Repository,
txn: &mut T,
channel: &mut ChannelRef<T>,
to_download: Vec<Hash>,
do_apply: bool,
) -> Result<(), anyhow::Error> {
let (mut send, mut recv) = tokio::sync::mpsc::channel(100);
let mut change_path_ = repo.changes_dir.clone();
let to_download_ = to_download.clone();
let mut self_ = std::mem::replace(self, RemoteRepo::None);
let t = tokio::spawn(async move {
let mut hashes = Vec::new();
for h in to_download_.iter() {
if self_
.start_change_download(*h, &mut change_path_, false)
.await?
{
hashes.push(*h);
}
}
debug!("hashes = {:?}", hashes);
self_
.wait_downloads(&change_path_, &hashes, &mut send)
.await?;
Ok(self_)
});
let mut ws = libanu::ApplyWorkspace::new();
let mut change_path = repo.changes_dir.clone();
for h in to_download.iter() {
libanu::changestore::filesystem::push_filename(&mut change_path, &h);
debug!("change_path = {:?}", change_path);
while std::fs::metadata(&change_path).is_err() {
debug!("waiting");
let r = recv.recv().await;
debug!("r = {:?}", r);
if r.is_none() {
break;
}
}
libanu::changestore::filesystem::pop_filename(&mut change_path);
if do_apply {
println!("Applying {:?}", h.to_base32());
debug!("applying {:?}", h);
txn.apply_change_ws(&repo.changes, channel, *h, &mut ws)?;
} else {
debug!("not applying {:?}", h)
}
}
std::mem::drop(recv);
debug!("waiting for spawned process");
let r: Result<_, anyhow::Error> = t.await?;
debug!("done");
*self = r?;
Ok(())
}
pub async fn clone_tag<T: MutTxnTExt + TxnTExt>(
&mut self,
repo: &mut Repository,
txn: &mut T,
channel: &mut ChannelRef<T>,
tag: &[Hash],
) -> Result<(), anyhow::Error> {
let (mut send_signal, mut recv_signal) = tokio::sync::mpsc::channel(100);
let (mut send_hash, mut recv_hash) = tokio::sync::mpsc::channel(100);
let mut change_path_ = repo.changes_dir.clone();
let mut self_ = std::mem::replace(self, RemoteRepo::None);
let t = tokio::spawn(async move {
let mut hashes = Vec::new();
while let Some(hash) = recv_hash.recv().await {
if self_
.start_change_download(hash, &mut change_path_, false)
.await?
{
hashes.push(hash);
}
}
debug!("hashes = {:?}", hashes);
self_
.wait_downloads(&change_path_, &hashes, &mut send_signal)
.await?;
Ok(self_)
});
for &h in tag.iter() {
send_hash.send(h).await?;
}
let mut change_path = repo.changes_dir.clone();
let mut hashes = Vec::new();
while let Some(hash) = recv_signal.recv().await {
libanu::changestore::filesystem::push_filename(&mut change_path, &hash);
std::fs::create_dir_all(change_path.parent().unwrap())?;
use libanu::changestore::ChangeStore;
hashes.push(hash);
for dep in repo.changes.get_dependencies(&hash)? {
let dep: libanu::pristine::Hash = dep;
send_hash.send(dep).await?;
}
libanu::changestore::filesystem::pop_filename(&mut change_path);
}
std::mem::drop(recv_signal);
std::mem::drop(send_hash);
let mut ws = libanu::ApplyWorkspace::new();
while let Some(hash) = hashes.pop() {
txn.apply_change_ws(&repo.changes, channel, hash, &mut ws)?;
}
let r: Result<_, anyhow::Error> = t.await?;
*self = r?;
Ok(())
}
pub async fn clone_state<T: MutTxnTExt + TxnTExt>(
&mut self,
repo: &mut Repository,
txn: &mut T,
channel: &mut ChannelRef<T>,
state: Merkle,
lazy: bool,
) -> Result<(), anyhow::Error> {
self.update_changelist(txn, &[]).await?;
let name = self.name();
let remote = txn.open_or_create_remote(name).unwrap();
if let RemoteRepo::Ssh(ref mut s) = self {
s.clone_channel(repo, txn, channel, lazy).await?;
let mut to_unrecord = Vec::new();
let mut found = false;
for (n, (h, s)) in txn.iter_rev_remote(&remote.borrow().remote, None) {
debug!("{:?} {:?} {:?}", n, h, s);
if s == state {
found = true;
break;
}
to_unrecord.push(h);
}
if !found {
return Err((Error::StateNotFound { state }).into());
}
self.pull(repo, txn, channel, to_unrecord.clone(), false)
.await?;
for unrec in to_unrecord.iter() {
txn.unrecord(&repo.changes, channel, unrec)?;
}
return Ok(());
}
let mut to_pull = Vec::new();
let mut found = false;
for (n, (h, s)) in txn.iter_remote(&remote.borrow().remote, 0) {
debug!("{:?} {:?} {:?}", n, h, s);
to_pull.push(h);
if s == state {
found = true;
break;
}
}
if !found {
return Err((Error::StateNotFound { state }).into());
}
self.pull(repo, txn, channel, to_pull, true).await?;
Ok(())
}
pub async fn complete_changes<T: MutTxnTExt + TxnTExt>(
&mut self,
repo: &crate::repository::Repository,
txn: &T,
local_channel: &mut ChannelRef<T>,
changes: &[Hash],
full: bool,
) -> Result<(), anyhow::Error> {
use libanu::changestore::ChangeStore;
let (mut send_hash, mut recv_hash) = tokio::sync::mpsc::channel(100);
let (mut send_sig, mut recv_sig) = tokio::sync::mpsc::channel(100);
let mut self_ = std::mem::replace(self, RemoteRepo::None);
let mut changes_dir = repo.changes_dir.clone();
let t = tokio::spawn(async move {
let mut hashes = Vec::new();
while let Some(h) = recv_hash.recv().await {
debug!("downloading full patch: {:?}", h);
if self_
.start_change_download(h, &mut changes_dir, true)
.await?
{
debug!("push");
hashes.push(h);
}
debug!("done");
}
debug!("waiting");
self_
.wait_downloads(&changes_dir, &hashes, &mut send_sig)
.await?;
let result: Result<_, anyhow::Error> = Ok(self_);
result
});
for c in changes {
if repo.changes.has_contents(*c, txn.get_internal(*c)) {
debug!("has contents {:?}", c);
continue;
}
if full {
debug!("sending send_hash");
send_hash.send(*c).await?;
debug!("sent");
continue;
}
let change = if let Some(i) = txn.get_internal(*c) {
i
} else {
continue;
};
// Check if at least one non-empty vertex from c is still alive.
let v = libanu::pristine::Vertex {
change,
start: libanu::pristine::ChangePosition(0),
end: libanu::pristine::ChangePosition(0),
};
let channel = local_channel.borrow();
for (v_, e) in txn.iter_graph(&channel.graph, v, None) {
if v_.change < change {
continue;
} else if v_.change > change {
break;
}
if e.flag.contains(libanu::pristine::EdgeFlags::PARENT)
&& !e.flag.contains(libanu::pristine::EdgeFlags::DELETED)
{
// Alive!
debug!("sending alive");
send_hash.send(*c).await?;
debug!("sent");
break;
}
}
}
debug!("dropping send_hash");
std::mem::drop(send_hash);
while recv_sig.recv().await.is_some() {}
*self = t.await??;
Ok(())
}
pub async fn clone_channel<T: MutTxnTExt + TxnTExt>(
&mut self,
repo: &mut Repository,
txn: &mut T,
local_channel: &mut ChannelRef<T>,
lazy: bool,
path: &[String],
) -> Result<(), anyhow::Error> {
if path.is_empty() {
match *self {
RemoteRepo::Ssh(ref mut s) => {
return s.clone_channel(repo, txn, local_channel, lazy).await
}
_ => {}
}
}
let remote_changes = self.update_changelist(txn, path).await?;
let mut pullable = Vec::new();
for (_, (h, _)) in txn.iter_remote(&remote_changes.borrow().remote, 0) {
pullable.push(h)
}
// let pullable = self.pullable(txn, local_channel, path).await?;
self.pull(repo, txn, local_channel, pullable, true).await
}
}
fn parse_line(data: &str) -> Result<(u64, Hash, Merkle), anyhow::Error> {
debug!("data = {:?}", data);
let mut it = data.split('.');
let n = if let Some(n) = it.next().and_then(|n| n.parse().ok()) {
n
} else {
return Err((Error::ProtocolError {
line: data.as_bytes().to_vec(),
})
.into());
};
debug!("n = {:?}", n);
let h = if let Some(h) = it.next().and_then(|h| Hash::from_base32(h.as_bytes())) {
h
} else {
return Err((Error::ProtocolError {
line: data.as_bytes().to_vec(),
})
.into());
};
debug!("h = {:?}", h);
let m = if let Some(m) = it.next().and_then(|m| {
debug!("m = {:?}", m);
Merkle::from_base32(m.as_bytes())
}) {
m
} else {
return Err((Error::ProtocolError {
line: data.as_bytes().to_vec(),
})
.into());
};
debug!("m = {:?}", m);
if it.next().is_some() {
return Err((Error::ProtocolError {
line: data.as_bytes().to_vec(),
})
.into());
}
Ok((n, h, m))
}