(import
  (fetchTarball {
    url = "https://github.com/edolstra/flake-compat/archive/c71d063a2fce96001c7ef99b301dfbd0c29ebde1.tar.gz";
    sha256 = "0vnbhqf0lc4mf2zmzqbfv6kqj9raijxz8xfaimxihz3c6s5gma2x";
  })
  { src = ./.; }).shellNix.default

edition = "2018"

#![recursion_limit = "256"]
extern crate proc_macro;
extern crate proc_macro2;
#[macro_use]
extern crate quote;
use proc_macro::TokenStream;
use proc_macro2::*;
use std::iter::FromIterator;
fn name_capital(name: &str) -> String {
    name.chars()
        .enumerate()
        .map(|(i, s)| {
            if i == 0 {
                s.to_uppercase().next().unwrap()
            } else {
                s
            }
        })
        .collect()
}
#[proc_macro]
pub fn table(input: proc_macro::TokenStream) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    assert!(input_iter.next().is_none());
    let name_capital = syn::Ident::new(&name_capital(&name), Span::call_site());
    proc_macro::TokenStream::from(quote! {
        type #name_capital;
    })
}
#[proc_macro]
pub fn sanakirja_table_get(input: proc_macro::TokenStream) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let name_get = syn::Ident::new(&format!("get_{}", name), Span::call_site());
    let name = syn::Ident::new(&name, Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let error = next(&mut input_iter);
    let error = if error.is_empty() {
        quote! { Error }
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    let pre_ = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let pre = if !pre_.is_empty() {
        quote! {
            let (key, value) = #pre_;
        }
    } else {
        quote! {}
    };
    let post_ = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let post = if post_.is_empty() {
        quote! { if let Ok(x) = self.txn.get(&self.#name, key, value) {
            Ok(x)
        } else {
            Err(TxnErr(SanakirjaError::PristineCorrupt))
        }
        }
    } else {
        quote! { if let Ok(x) = self.txn.get(&self.#name, key, value) {
            Ok(x . #post_)
        } else {
            Err(TxnErr(SanakirjaError::PristineCorrupt))
        }}
    };
    proc_macro::TokenStream::from(quote! {
        fn #name_get <'txn> (&'txn self, key: #key, value: Option<#value>) -> Result<Option<#value>, TxnErr<Self::#error>> {
            use ::sanakirja::Transaction;
            #pre
            #post
        }
    })
}
#[proc_macro]
pub fn sanakirja_get(input: proc_macro::TokenStream) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let name_capital = syn::Ident::new(&name_capital(&name), Span::call_site());
    let name_get = syn::Ident::new(&format!("get_{}", name), Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let error = next(&mut input_iter);
    let error = if error.is_empty() {
        quote! { Error }
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    assert!(input_iter.next().is_none());
    proc_macro::TokenStream::from(quote! {
        fn #name_get(&self, db: &Self::#name_capital, key: #key, value: Option<#value>) -> Result<Option<#value>, TxnErr<Self::#error>> {
            use ::sanakirja::Transaction;
            if let Ok(x) = self.txn.get(db, key, value) {
                Ok(x)
            } else {
                Err(TxnErr(SanakirjaError::PristineCorrupt))
            }
        }
    })
}
#[proc_macro]
pub fn table_get(input: proc_macro::TokenStream) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let name_get = syn::Ident::new(&format!("get_{}", name), Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let error = next(&mut input_iter);
    let error = if error.is_empty() {
        quote! { Error }
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    assert!(input_iter.next().is_none());
    proc_macro::TokenStream::from(quote! {
        fn #name_get<'txn>(&'txn self, key: #key, value: Option<#value>) -> Result<Option<#value>, TxnErr<Self::#error>>;
    })
}
#[proc_macro]
pub fn get(input: proc_macro::TokenStream) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let name_capital = syn::Ident::new(&name_capital(&name), Span::call_site());
    let name_get = syn::Ident::new(&format!("get_{}", name), Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let error = next(&mut input_iter);
    let error = if error.is_empty() {
        quote! { Error }
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    assert!(input_iter.next().is_none());
    proc_macro::TokenStream::from(quote! {
        fn #name_get<'txn>(&'txn self, db: &Self::#name_capital, key: #key, value: Option<#value>) -> Result<Option<#value>, TxnErr<Self::#error>>;
    })
}
fn next(input_iter: &mut proc_macro2::token_stream::IntoIter) -> Vec<TokenTree> {
    let mut result = Vec::new();
    let mut is_first = true;
    loop {
        match input_iter.next() {
            Some(TokenTree::Punct(p)) => {
                if p.as_char() == ',' {
                    if !is_first {
                        return result;
                    }
                } else {
                    result.push(TokenTree::Punct(p))
                }
            }
            Some(e) => result.push(e),
            None => return result,
        }
        is_first = false
    }
}
#[proc_macro]
pub fn cursor(input: proc_macro::TokenStream) -> TokenStream {
    cursor_(input, false, false, false)
}
#[proc_macro]
pub fn cursor_ref(input: proc_macro::TokenStream) -> TokenStream {
    cursor_(input, false, false, true)
}
#[proc_macro]
pub fn iter(input: proc_macro::TokenStream) -> TokenStream {
    cursor_(input, false, true, false)
}
#[proc_macro]
pub fn rev_cursor(input: proc_macro::TokenStream) -> TokenStream {
    cursor_(input, true, false, false)
}
fn cursor_(input: proc_macro::TokenStream, rev: bool, iter: bool, borrow: bool) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let capital = name_capital(&name);
    let cursor_name = syn::Ident::new(&format!("{}Cursor", capital,), Span::call_site());
    let name_capital = syn::Ident::new(&name_capital(&name), Span::call_site());
    let name_iter = syn::Ident::new(&format!("iter_{}", name), Span::call_site());
    let name_next = syn::Ident::new(&format!("cursor_{}_next", name), Span::call_site());
    let name_prev = syn::Ident::new(&format!("cursor_{}_prev", name), Span::call_site());
    let name_cursor = syn::Ident::new(
        &format!("{}cursor_{}", if rev { "rev_" } else { "" }, name),
        Span::call_site(),
    );
    let name_cursor_ref = syn::Ident::new(
        &format!("{}cursor_{}_ref", if rev { "rev_" } else { "" }, name),
        Span::call_site(),
    );
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let error = next(&mut input_iter);
    let error = if error.is_empty() {
        quote! { GraphError }
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    let cursor_type = if rev {
        quote! {
            Result<crate::pristine::RevCursor<Self, &'txn Self, Self::#cursor_name, #key, #value>, TxnErr<Self::#error>>
        }
    } else {
        quote! {
            Result<crate::pristine::Cursor<Self, &'txn Self, Self::#cursor_name, #key, #value>, TxnErr<Self::#error>>
        }
    };
    let def = if rev {
        quote! {}
    } else {
        quote! {
            type #cursor_name;
            fn #name_next <'txn> (
                &'txn self,
                cursor: &mut Self::#cursor_name,
            ) -> Result<Option<(#key, #value)>, TxnErr<Self::#error>>;
            fn #name_prev <'txn> (
                &'txn self,
                cursor: &mut Self::#cursor_name,
            ) -> Result<Option<(#key, #value)>, TxnErr<Self::#error>>;
        }
    };
    let borrow = if borrow {
        quote! {
        fn #name_cursor_ref<RT: std::ops::Deref<Target = Self>>(
            txn: RT,
            db: &Self::#name_capital,
            pos: Option<(#key, Option<#value>)>,
        ) -> Result<crate::pristine::Cursor<Self, RT, Self::#cursor_name, #key, #value>, TxnErr<Self::#error>>;
        }
    } else {
        quote! {}
    };
    let iter = if !iter {
        quote! {}
    } else {
        quote! {
            fn #name_iter <'txn> (
                &'txn self,
                k: #key,
                v: Option<#value>
            ) -> #cursor_type;
        }
    };
    assert!(input_iter.next().is_none());
    proc_macro::TokenStream::from(quote! {
        #def
        fn #name_cursor<'txn>(
            &'txn self,
            db: &Self::#name_capital,
            pos: Option<(#key, Option<#value>)>,
        ) -> #cursor_type;
        #borrow
        #iter
    })
}
#[proc_macro]
pub fn sanakirja_cursor(input: proc_macro::TokenStream) -> TokenStream {
    sanakirja_cursor_(input, false, false, false)
}
#[proc_macro]
pub fn sanakirja_cursor_ref(input: proc_macro::TokenStream) -> TokenStream {
    sanakirja_cursor_(input, false, false, true)
}
#[proc_macro]
pub fn sanakirja_iter(input: proc_macro::TokenStream) -> TokenStream {
    sanakirja_cursor_(input, false, true, false)
}
#[proc_macro]
pub fn sanakirja_rev_cursor(input: proc_macro::TokenStream) -> TokenStream {
    sanakirja_cursor_(input, true, false, false)
}
fn sanakirja_cursor_(
    input: proc_macro::TokenStream,
    rev: bool,
    iter: bool,
    borrow: bool,
) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let cursor_name = syn::Ident::new(
        &format!("{}Cursor", name_capital(&name),),
        Span::call_site(),
    );
    let name_capital = syn::Ident::new(&name_capital(&name), Span::call_site());
    let name_next = syn::Ident::new(&format!("cursor_{}_next", name), Span::call_site());
    let name_prev = syn::Ident::new(&format!("cursor_{}_prev", name), Span::call_site());
    let name_cursor = syn::Ident::new(
        &format!("{}cursor_{}", if rev { "rev_" } else { "" }, name),
        Span::call_site(),
    );
    let name_cursor_ref = syn::Ident::new(
        &format!("{}cursor_{}_ref", if rev { "rev_" } else { "" }, name),
        Span::call_site(),
    );
    let name_iter = syn::Ident::new(
        &format!("{}iter_{}", if rev { "rev_" } else { "" }, name),
        Span::call_site(),
    );
    let name = syn::Ident::new(&name, Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let pre = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let post = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let pre_init = if !pre.is_empty() {
        quote! { let pos = #pre; }
    } else {
        quote! {}
    };
    let post = if !post.is_empty() {
        quote! { . #post }
    } else {
        quote! {}
    };
    let iter = if iter {
        quote! {
            fn #name_iter <'txn> (
                &'txn self,
                k: #key,
                v: Option<#value>
            ) -> Result<Cursor<Self, &'txn Self, Self::#cursor_name, #key, #value>, TxnErr<SanakirjaError>> {
                self.#name_cursor(&self.#name, Some((k, v)))
            }
        }
    } else {
        quote! {}
    };
    let borrow = if borrow {
        quote! {
            fn #name_cursor_ref <RT: std::ops::Deref<Target = Self>> (
                txn: RT,
                db: &Self::#name_capital,
                pos: Option<(#key, Option<#value>)>,
            ) -> Result<Cursor<Self, RT, Self::#cursor_name, #key, #value>, TxnErr<SanakirjaError>> {
                #pre_init
                if let Ok((cursor, _)) = txn.txn.set_cursors(&db, pos) {
                    Ok(Cursor {
                        cursor,
                        txn,
                        marker: std::marker::PhantomData,
                    })
                } else {
                    Err(TxnErr(SanakirjaError::PristineCorrupt))
                }
            }
        }
    } else {
        quote! {}
    };
    proc_macro::TokenStream::from(if rev {
        quote! {
            fn #name_cursor<'txn>(
                &'txn self,
                db: &Self::#name_capital,
                pos: Option<(#key, Option<#value>)>,
            ) -> Result<super::RevCursor<Self, &'txn Self, Self::#cursor_name, #key, #value>, TxnErr<SanakirjaError>> {
                #pre_init
                let mut cursor = if pos.is_some() {
                    if let Ok((x, _)) = self.txn.set_cursors(&db, pos) {
                        x
                    } else {
                        return Err(TxnErr(SanakirjaError::PristineCorrupt))
                    }
                } else if let Ok(x) = self.txn.set_cursors_last(&db) {
                    x
                } else {
                    return Err(TxnErr(SanakirjaError::PristineCorrupt))
                };
                Ok(super::RevCursor {
                    cursor,
                    txn: self,
                    marker: std::marker::PhantomData,
                })
            }
        }
    } else {
        quote! {
            type #cursor_name = ::sanakirja::Cursor;
            fn #name_cursor<'txn>(
                &'txn self,
                db: &Self::#name_capital,
                pos: Option<(#key, Option<#value>)>,
            ) -> Result<Cursor<Self, &'txn Self, Self::#cursor_name, #key, #value>, TxnErr<SanakirjaError>> {
                #pre_init
                let mut cursor = if let Ok((x, _)) = self.txn.set_cursors(&db, pos) {
                    x
                } else {
                    return Err(TxnErr(SanakirjaError::PristineCorrupt))
                };
                Ok(Cursor {
                    cursor,
                    txn: self,
                    marker: std::marker::PhantomData,
                })
            }
            #borrow
            fn #name_next <'txn> (
                &'txn self,
                cursor: &mut Self::#cursor_name,
            ) -> Result<Option<(#key, #value)>, TxnErr<SanakirjaError>> {
                let x = if let Ok(x) = unsafe { ::sanakirja::next(&self.txn, cursor) } {
                    x
                } else {
                    return Err(TxnErr(SanakirjaError::PristineCorrupt))
                };
                Ok(x #post)
            }
            fn #name_prev <'txn> (
                &'txn self,
                cursor: &mut Self::#cursor_name,
            ) -> Result<Option<(#key, #value)>, TxnErr<SanakirjaError>> {
                let x = if let Ok(x) = unsafe { ::sanakirja::prev(&self.txn, cursor) } {
                    x
                } else {
                    return Err(TxnErr(SanakirjaError::PristineCorrupt))
                };
                Ok(x #post)
            }
            #iter
        }
    })
}
#[proc_macro]
pub fn initialized_cursor(input: proc_macro::TokenStream) -> TokenStream {
    initialized_cursor_(input, false)
}
#[proc_macro]
pub fn initialized_rev_cursor(input: proc_macro::TokenStream) -> TokenStream {
    initialized_cursor_(input, true)
}
fn initialized_cursor_(input: proc_macro::TokenStream, rev: bool) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let cursor_name = syn::Ident::new(
        &format!("{}Cursor", name_capital(&name),),
        Span::call_site(),
    );
    let name_next = syn::Ident::new(&format!("cursor_{}_next", name), Span::call_site());
    let name_prev = syn::Ident::new(&format!("cursor_{}_prev", name), Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let txnt = next(&mut input_iter);
    let txnt: proc_macro2::TokenStream = if txnt.is_empty() {
        proc_macro2::TokenStream::from(quote! { TxnT })
    } else {
        proc_macro2::TokenStream::from_iter(txnt.into_iter())
    };
    let error = next(&mut input_iter);
    let error: proc_macro2::TokenStream = if error.is_empty() {
        proc_macro2::TokenStream::from(quote! { GraphError })
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    assert!(input_iter.next().is_none());
    if rev {
        proc_macro::TokenStream::from(quote! {
            impl<T: #txnt, RT: std::ops::Deref<Target = T>> Iterator for crate::pristine::RevCursor<T, RT, T::#cursor_name, #key, #value>
            {
                type Item = Result<(#key, #value), TxnErr<T::#error>>;
                fn next(&mut self) -> Option<Self::Item> {
                    match self.txn.#name_prev(&mut self.cursor) {
                        Ok(Some(x)) => Some(Ok(x)),
                        Ok(None) => None,
                        Err(e) => Some(Err(e)),
                    }
                }
            }
        })
    } else {
        proc_macro::TokenStream::from(quote! {
            impl<T: #txnt, RT: std::ops::Deref<Target = T>>
                crate::pristine::Cursor<T, RT, T::#cursor_name, #key, #value>
            {
                pub fn prev(&mut self) -> Option<Result<(#key, #value), TxnErr<T::#error>>> {
                    match self.txn.#name_prev(&mut self.cursor) {
                        Ok(Some(x)) => Some(Ok(x)),
                        Ok(None) => None,
                        Err(e) => Some(Err(e)),
                    }
                }
            }
            impl<T: #txnt, RT: std::ops::Deref<Target = T>> Iterator for crate::pristine::Cursor<T, RT, T::#cursor_name, #key, #value>
            {
                type Item = Result<(#key, #value), TxnErr<T::#error>>;
                fn next(&mut self) -> Option<Self::Item> {
                    match self.txn.#name_next(&mut self.cursor) {
                        Ok(Some(x)) => Some(Ok(x)),
                        Ok(None) => None,
                        Err(e) => Some(Err(e)),
                    }
                }
            }
        })
    }
}
#[proc_macro]
pub fn put_del(input: proc_macro::TokenStream) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let put = syn::Ident::new(&format!("put_{}", name), Span::call_site());
    let del = syn::Ident::new(&format!("del_{}", name), Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let error = next(&mut input_iter);
    let error = if error.is_empty() {
        quote! { Error }
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    assert!(input_iter.next().is_none());
    proc_macro::TokenStream::from(quote! {
        fn #put(
            &mut self,
            k: #key,
            e: #value,
        ) -> Result<bool, TxnErr<Self::#error>>;
        fn #del(
            &mut self,
            k: #key,
            e: Option<#value>,
        ) -> Result<bool, TxnErr<Self::#error>>;
    })
}
#[proc_macro]
pub fn sanakirja_put_del(input: proc_macro::TokenStream) -> TokenStream {
    let input = proc_macro2::TokenStream::from(input);
    let mut input_iter = input.into_iter();
    let name = match input_iter.next() {
        Some(TokenTree::Ident(id)) => id.to_string(),
        _ => panic!("txn_table: first argument not an identifier"),
    };
    let put = syn::Ident::new(&format!("put_{}", name), Span::call_site());
    let del = syn::Ident::new(&format!("del_{}", name), Span::call_site());
    let name = syn::Ident::new(&name, Span::call_site());
    let key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let error = next(&mut input_iter);
    let error = if error.is_empty() {
        quote! { Error }
    } else {
        proc_macro2::TokenStream::from_iter(error.into_iter())
    };
    let pre_key = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    let pre_value = proc_macro2::TokenStream::from_iter(next(&mut input_iter).into_iter());
    assert!(input_iter.next().is_none());
    if pre_key.is_empty() {
        proc_macro::TokenStream::from(quote! {
            fn #put(
                &mut self,
                k: #key,
                v: #value,
            ) -> Result<bool, TxnErr<Self::#error>> {
                Ok(self.txn.put(&mut self.rng, &mut self.#name, k, v).map_err(TxnErr)?)
            }
            fn #del(
                &mut self,
                k: #key,
                v: Option<#value>,
            ) -> Result<bool, TxnErr<Self::#error>> {
                Ok(self.txn.del(&mut self.rng, &mut self.#name, k, v).map_err(TxnErr)?)
            }
        })
    } else {
        proc_macro::TokenStream::from(quote! {
            fn #put(
                &mut self,
                k: #key,
                v: #value,
            ) -> Result<bool, TxnErr<Self::#error>> {
                let k = #pre_key;
                let v = #pre_value;
                Ok(self.txn.put(&mut self.rng, &mut self.#name, k, v).map_err(TxnErr)?)
            }
            fn #del(
                &mut self,
                k: #key,
                v: Option<#value>,
            ) -> Result<bool, TxnErr<Self::#error>> {
                let k = #pre_key;
                let v = v.map(|v| #pre_value);
                Ok(self.txn.del(&mut self.rng, &mut self.#name, k, v).map_err(TxnErr)?)
            }
        })
    }
}

[package]
name = "pijul-macros"
description = "Macros used to write libpijul."
version = "0.3.0"
authors = ["Pierre-Étienne Meunier <pmeunier@mailbox.org>"]
edition = "2018"
repository = "https://nest.pijul.com/pijul/pijul"
license = "GPL-2.0"
include = [ "Cargo.toml", "src/lib.rs" ]
[lib]
proc-macro = true
[dependencies]
syn = "1.0"
quote = "1.0"
proc-macro2 = "1.0"
regex = "1.4"

use std::io::Write;
use std::path::PathBuf;
use anyhow::bail;
use libpijul::DOT_DIR;
use log::debug;
use crate::{config, current_dir};
pub struct Repository {
    pub pristine: libpijul::pristine::sanakirja::Pristine,
    pub changes: libpijul::changestore::filesystem::FileSystem,
    pub working_copy: libpijul::working_copy::filesystem::FileSystem,
    pub config: config::Config,
    pub path: PathBuf,
    pub changes_dir: PathBuf,
}
pub const PRISTINE_DIR: &str = "pristine";
pub const CHANGES_DIR: &str = "changes";
pub const CONFIG_FILE: &str = "config";
impl Repository {
    pub fn save_config(&self) -> Result<(), anyhow::Error> {
        let config = toml::to_string(&self.config)?;
        let mut file = std::fs::File::create(&self.path.join(DOT_DIR).join(CONFIG_FILE))?;
        file.write_all(config.as_bytes())?;
        Ok(())
    }
    fn find_root_(cur: Option<PathBuf>, dot_dir: &str) -> Result<PathBuf, anyhow::Error> {
        let mut cur = if let Some(cur) = cur {
            cur
        } else {
            current_dir()?
        };
        cur.push(dot_dir);
        loop {
            debug!("{:?}", cur);
            if std::fs::metadata(&cur).is_err() {
                cur.pop();
                if cur.pop() {
                    cur.push(DOT_DIR);
                } else {
                    bail!("No Pijul repository found")
                }
            } else {
                break;
            }
        }
        Ok(cur)
    }
    pub fn find_root(cur: Option<PathBuf>) -> Result<Self, anyhow::Error> {
        Self::find_root_with_dot_dir(cur, DOT_DIR)
    }
    pub fn find_root_with_dot_dir(
        cur: Option<PathBuf>,
        dot_dir: &str,
    ) -> Result<Self, anyhow::Error> {
        let cur = Self::find_root_(cur, dot_dir)?;
        let mut pristine_dir = cur.clone();
        pristine_dir.push(PRISTINE_DIR);
        let mut changes_dir = cur.clone();
        changes_dir.push(CHANGES_DIR);
        let mut working_copy_dir = cur.clone();
        working_copy_dir.pop();
        let config_path = cur.join(CONFIG_FILE);
        let config = if let Ok(config) = std::fs::read(&config_path) {
            if let Ok(toml) = toml::from_slice(&config) {
                toml
            } else {
                bail!("Could not read configuration file at {:?}", config_path)
            }
        } else {
            config::Config::default()
        };
        Ok(Repository {
            pristine: libpijul::pristine::sanakirja::Pristine::new(&pristine_dir)?,
            working_copy: libpijul::working_copy::filesystem::FileSystem::from_root(
                &working_copy_dir,
            ),
            changes: libpijul::changestore::filesystem::FileSystem::from_root(&working_copy_dir),
            config,
            path: working_copy_dir,
            changes_dir,
        })
    }
    pub fn init(path: Option<std::path::PathBuf>) -> Result<Self, anyhow::Error> {
        let cur = if let Some(path) = path {
            path
        } else {
            current_dir()?
        };
        let mut pristine_dir = cur.clone();
        pristine_dir.push(DOT_DIR);
        pristine_dir.push(PRISTINE_DIR);
        if std::fs::metadata(&pristine_dir).is_err() {
            std::fs::create_dir_all(&pristine_dir)?;
            let mut changes_dir = cur.clone();
            changes_dir.push(DOT_DIR);
            changes_dir.push(CHANGES_DIR);
            Ok(Repository {
                pristine: libpijul::pristine::sanakirja::Pristine::new(&pristine_dir)?,
                working_copy: libpijul::working_copy::filesystem::FileSystem::from_root(&cur),
                changes: libpijul::changestore::filesystem::FileSystem::from_root(&cur),
                config: config::Config::default(),
                path: cur,
                changes_dir,
            })
        } else {
            bail!("Already in a repository")
        }
    }
}

use std::collections::{HashMap, HashSet};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use anyhow::bail;
use byteorder::{BigEndian, ReadBytesExt};
use indicatif::ProgressBar;
use lazy_static::lazy_static;
use libpijul::pristine::Position;
use libpijul::{Base32, ChannelRef, Hash, Merkle, MutTxnT, MutTxnTExt, TxnTExt};
use log::{debug, error};
use regex::Regex;
use thrussh::client::Session;
use tokio::sync::Mutex;
use super::{parse_line, RemoteRef};
use crate::repository::Repository;
pub struct Ssh {
    pub h: thrussh::client::Handle<SshClient>,
    pub c: thrussh::client::Channel,
    pub channel: String,
    pub remote_cmd: String,
    pub path: String,
    pub is_running: bool,
    pub name: String,
    state: Arc<Mutex<State>>,
    has_errors: Arc<Mutex<bool>>,
}
lazy_static! {
    static ref ADDRESS: Regex = Regex::new(
        r#"(ssh://)?((?P<user>[^@]+)@)?((?P<host>(\[([^\]]+)\])|([^:/]+)))((:(?P<port>\d+)/)|:|/)(?P<path>.+)"#
    )
        .unwrap();
    static ref PROGRESS: Arc<Mutex<Option<ProgressBar>>> = Arc::new(Mutex::new(None));
}
#[derive(Debug)]
pub struct Remote<'a> {
    addr: &'a str,
    host: &'a str,
    path: &'a str,
    config: thrussh_config::Config,
}
pub fn ssh_remote(addr: &str) -> Option<Remote> {
    let cap = ADDRESS.captures(addr)?;
    debug!("ssh_remote: {:?}", cap);
    let host = cap.name("host").unwrap().as_str();
    let mut config =
        thrussh_config::parse_home(&host).unwrap_or(thrussh_config::Config::default(host));
    if let Some(port) = cap.name("port").map(|x| x.as_str().parse().unwrap()) {
        config.port = port
    }
    if let Some(u) = cap.name("user") {
        config.user.clear();
        config.user.push_str(u.as_str());
    }
    let path = cap.name("path").unwrap().as_str();
    Some(Remote {
        addr,
        host,
        path,
        config,
    })
}
impl<'a> Remote<'a> {
    pub async fn connect(&mut self, name: &str, channel: &str) -> Result<Ssh, anyhow::Error> {
        let mut home = dirs_next::home_dir().unwrap();
        home.push(".ssh");
        home.push("known_hosts");
        let state = Arc::new(Mutex::new(State::None));
        let has_errors = Arc::new(Mutex::new(false));
        let client = SshClient {
            addr: self.config.host_name.clone(),
            known_hosts: home,
            last_window_adjustment: SystemTime::now(),
            state: state.clone(),
            has_errors: has_errors.clone(),
        };
        let stream = self.config.stream().await?;
        let config = Arc::new(thrussh::client::Config::default());
        let mut h = thrussh::client::connect_stream(config, stream, client).await?;
        let mut key_path = dirs_next::home_dir().unwrap().join(".ssh");
        // First try agent auth
        let authenticated = self
            .auth_agent(&mut h, &mut key_path)
            .await
            .unwrap_or(false)
            || self.auth_pk(&mut h, &mut key_path).await
            || self.auth_password(&mut h).await?;
        if !authenticated {
            bail!("Not authenticated")
        }
        let c = h.channel_open_session().await?;
        let remote_cmd = if let Ok(cmd) = std::env::var("REMOTE_PIJUL") {
            cmd
        } else {
            "pijul".to_string()
        };
        Ok(Ssh {
            h,
            c,
            channel: channel.to_string(),
            remote_cmd,
            path: self.path.to_string(),
            is_running: false,
            name: name.to_string(),
            state,
            has_errors,
        })
    }
    async fn auth_agent(
        &self,
        h: &mut thrussh::client::Handle<SshClient>,
        key_path: &mut PathBuf,
    ) -> Result<bool, anyhow::Error> {
        let mut authenticated = false;
        let mut agent = thrussh_keys::agent::client::AgentClient::connect_env().await?;
        let identities = if let Some(ref file) = self.config.identity_file {
            key_path.push(file);
            key_path.set_extension("pub");
            let k = thrussh_keys::load_public_key(&key_path);
            key_path.pop();
            if let Ok(k) = k {
                vec![k]
            } else {
                return Ok(false);
            }
        } else {
            agent.request_identities().await?
        };
        debug!("identities = {:?}", identities);
        let mut agent = Some(agent);
        for key in identities {
            debug!("Trying key {:?}", key);
            debug!("fingerprint = {:?}", key.fingerprint());
            if let Some(a) = agent.take() {
                debug!("authenticate future");
                match h.authenticate_future(&self.config.user, key, a).await {
                    (a, Ok(auth)) => {
                        authenticated = auth;
                        agent = Some(a);
                    }
                    (a, Err(e)) => {
                        agent = Some(a);
                        debug!("not auth {:?}", e);
                        if let thrussh::AgentAuthError::Key(e) = e {
                            debug!("error: {:?}", e);
                            writeln!(std::io::stderr(), "Failed to sign with agent")?;
                        }
                    }
                }
            }
            if authenticated {
                return Ok(true);
            }
        }
        Ok(false)
    }
    async fn auth_pk(
        &self,
        h: &mut thrussh::client::Handle<SshClient>,
        key_path: &mut PathBuf,
    ) -> bool {
        let mut authenticated = false;
        let mut keys = Vec::new();
        if let Some(ref file) = self.config.identity_file {
            keys.push(file.as_str())
        } else {
            keys.push("id_ed25519");
            keys.push("id_rsa");
        }
        for k in keys.iter() {
            key_path.push(k);
            let k = if let Some(k) = load_secret_key(&key_path, k) {
                k
            } else {
                key_path.pop();
                continue;
            };
            if let Ok(auth) = h
                .authenticate_publickey(&self.config.user, Arc::new(k))
                .await
            {
                authenticated = auth
            }
            key_path.pop();
            if authenticated {
                return true;
            }
        }
        false
    }
    async fn auth_password(
        &self,
        h: &mut thrussh::client::Handle<SshClient>,
    ) -> Result<bool, thrussh::Error> {
        let pass = rpassword::read_password_from_tty(Some(&format!(
            "Password for {}@{}: ",
            self.config.user, self.config.host_name
        )))?;
        h.authenticate_password(self.config.user.to_string(), &pass)
            .await
    }
}
pub fn load_secret_key(key_path: &Path, k: &str) -> Option<thrussh_keys::key::KeyPair> {
    match thrussh_keys::load_secret_key(&key_path, None) {
        Ok(k) => Some(k),
        Err(e) => {
            if let thrussh_keys::Error::KeyIsEncrypted = e {
                let pass = if let Ok(pass) =
                    rpassword::read_password_from_tty(Some(&format!("Password for key {:?}: ", k)))
                {
                    pass
                } else {
                    return None;
                };
                if pass.is_empty() {
                    return None;
                }
                if let Ok(k) = thrussh_keys::load_secret_key(&key_path, Some(pass.as_bytes())) {
                    return Some(k);
                }
            }
            None
        }
    }
}
pub struct SshClient {
    addr: String,
    known_hosts: PathBuf,
    last_window_adjustment: SystemTime,
    state: Arc<Mutex<State>>,
    has_errors: Arc<Mutex<bool>>,
}
enum State {
    None,
    State {
        sender: Option<tokio::sync::oneshot::Sender<Option<(u64, Merkle)>>>,
    },
    Changes {
        sender: Option<tokio::sync::mpsc::Sender<Hash>>,
        remaining_len: usize,
        file: std::fs::File,
        path: PathBuf,
        hashes: Vec<libpijul::pristine::Hash>,
        current: usize,
    },
    Changelist {
        sender: tokio::sync::mpsc::Sender<Option<super::ListLine>>,
    },
    Channel {
        sender: tokio::sync::mpsc::Sender<Vec<u8>>,
    },
    Archive {
        sender: Option<tokio::sync::oneshot::Sender<u64>>,
        len: u64,
        conflicts: u64,
        len_n: u64,
        w: Box<dyn Write + Send>,
    },
}
type BoxFuture<T> = Pin<Box<dyn futures::future::Future<Output = T> + Send>>;
impl thrussh::client::Handler for SshClient {
    type Error = anyhow::Error;
    type FutureBool = futures::future::Ready<Result<(Self, bool), anyhow::Error>>;
    type FutureUnit = BoxFuture<Result<(Self, Session), anyhow::Error>>;
    fn finished_bool(self, b: bool) -> Self::FutureBool {
        futures::future::ready(Ok((self, b)))
    }
    fn finished(self, session: Session) -> Self::FutureUnit {
        Box::pin(async move { Ok((self, session)) })
    }
    fn check_server_key(
        self,
        server_public_key: &thrussh_keys::key::PublicKey,
    ) -> Self::FutureBool {
        let mut it = self.addr.split(':');
        let addr = it.next().unwrap();
        let port = it.next().unwrap_or("22").parse().unwrap();
        match thrussh_keys::check_known_hosts_path(addr, port, server_public_key, &self.known_hosts)
        {
            Ok(e) => {
                if e {
                    futures::future::ready(Ok((self, true)))
                } else {
                    match learn(addr, port, server_public_key) {
                        Ok(x) => futures::future::ready(Ok((self, x))),
                        Err(e) => futures::future::ready(Err(e)),
                    }
                }
            }
            Err(e) => {
                error!("Key changed for {:?}", self.addr);
                futures::future::ready(Err(e.into()))
            }
        }
    }
    fn adjust_window(&mut self, _channel: thrussh::ChannelId, target: u32) -> u32 {
        let elapsed = self.last_window_adjustment.elapsed().unwrap();
        self.last_window_adjustment = SystemTime::now();
        if target >= 10_000_000 {
            return target;
        }
        if elapsed < Duration::from_secs(2) {
            target * 2
        } else if elapsed > Duration::from_secs(8) {
            target / 2
        } else {
            target
        }
    }
    fn channel_eof(
        self,
        _channel: thrussh::ChannelId,
        session: thrussh::client::Session,
    ) -> Self::FutureUnit {
        Box::pin(async move {
            *self.state.lock().await = State::None;
            Ok((self, session))
        })
    }
    fn exit_status(
        self,
        _channel: thrussh::ChannelId,
        _status: u32,
        session: thrussh::client::Session,
    ) -> Self::FutureUnit {
        Box::pin(async move {
            *self.state.lock().await = State::None;
            *self.has_errors.lock().await = true;
            Ok((self, session))
        })
    }
    fn extended_data(
        self,
        channel: thrussh::ChannelId,
        ext: u32,
        data: &[u8],
        session: thrussh::client::Session,
    ) -> Self::FutureUnit {
        debug!("extended data {:?}, {:?}", std::str::from_utf8(data), ext);
        if ext == 0 {
            self.data(channel, data, session)
        } else {
            let data = data.to_vec();
            Box::pin(async move {
                *PROGRESS.lock().await = None;
                *super::SPINNER.lock().await = None;
                *self.has_errors.lock().await = true;
                let stderr = std::io::stderr();
                let mut handle = stderr.lock();
                handle.write_all(&data)?;
                Ok((self, session))
            })
        }
    }
    fn data(
        self,
        channel: thrussh::ChannelId,
        data: &[u8],
        session: thrussh::client::Session,
    ) -> Self::FutureUnit {
        debug!("data {:?} {:?}", channel, data.len());
        let data = data.to_vec();
        Box::pin(async move {
            match *self.state.lock().await {
                State::State { ref mut sender } => {
                    debug!("state: State");
                    if let Some(sender) = sender.take() {
                        // If we can't parse `data` (for example if the
                        // remote returns the standard "-\n"), this
                        // returns None.
                        let mut s = std::str::from_utf8(&data).unwrap().split(' ');
                        debug!("s = {:?}", s);
                        if let (Some(n), Some(m)) = (s.next(), s.next()) {
                            let n = n.parse().unwrap();
                            sender
                                .send(Some((n, Merkle::from_base32(m.trim().as_bytes()).unwrap())))
                                .unwrap_or(());
                        } else {
                            sender.send(None).unwrap_or(());
                        }
                    }
                }
                State::Changes {
                    ref mut sender,
                    ref mut remaining_len,
                    ref mut file,
                    ref mut path,
                    ref hashes,
                    ref mut current,
                } => {
                    debug!("state changes");
                    let mut p = 0;
                    while p < data.len() {
                        if *remaining_len == 0 {
                            *remaining_len = (&data[p..]).read_u64::<BigEndian>().unwrap() as usize;
                            p += 8;
                            debug!("remaining_len = {:?}", remaining_len);
                        }
                        if data.len() >= p + *remaining_len {
                            file.write_all(&data[p..p + *remaining_len])?;
                            // We have enough data to write the
                            // file, write it and move to the next
                            // file.
                            p += *remaining_len;
                            *remaining_len = 0;
                            file.flush()?;
                            let mut final_path = path.clone();
                            final_path.set_extension("change");
                            debug!("moving {:?} to {:?}", path, final_path);
                            std::fs::rename(&path, &final_path)?;
                            debug!("sending");
                            if let Some(ref mut sender) = sender {
                                if sender.send(hashes[*current]).await.is_err() {
                                    break;
                                }
                            }
                            debug!("sent");
                            *current += 1;
                            if *current < hashes.len() {
                                // If we're still waiting for
                                // another change.
                                libpijul::changestore::filesystem::pop_filename(path);
                                libpijul::changestore::filesystem::push_filename(
                                    path,
                                    &hashes[*current],
                                );
                                std::fs::create_dir_all(&path.parent().unwrap())?;
                                path.set_extension("tmp");
                                debug!("creating file {:?}", path);
                                *file = std::fs::File::create(&path)?;
                            } else {
                                // Else, just finish.
                                debug!("dropping channel");
                                std::mem::drop(sender.take());
                                break;
                            }
                        } else {
                            // not enough data, we need more.
                            file.write_all(&data[p..])?;
                            file.flush()?;
                            *remaining_len -= data.len() - p;
                            debug!("need more data");
                            break;
                        }
                    }
                    debug!("finished, {:?} {:?}", p, data.len());
                }
                State::Changelist { ref mut sender } => {
                    debug!("state changelist");
                    if &data[..] == b"\n" {
                        debug!("log done");
                        sender.send(None).await.unwrap_or(())
                    } else if let Ok(data) = std::str::from_utf8(&data) {
                        for l in data.lines() {
                            if !l.is_empty() {
                                debug!("line = {:?}", l);
                                sender.send(parse_line(l).ok()).await.unwrap_or(())
                            } else {
                                sender.send(None).await.unwrap_or(());
                            }
                        }
                    }
                }
                State::Channel { ref mut sender } => {
                    debug!("state channel");
                    sender.send(data).await?
                }
                State::Archive {
                    ref mut sender,
                    ref mut w,
                    ref mut len,
                    ref mut len_n,
                    ref mut conflicts,
                } => {
                    debug!("state archive");
                    let mut off = 0;
                    while *len_n < 16 && off < data.len() {
                        if *len_n < 8 {
                            *len = (*len << 8) | (data[off] as u64);
                        } else {
                            *conflicts = (*conflicts << 8) | (data[off] as u64);
                        }
                        *len_n += 1;
                        off += 1;
                    }
                    if *len_n >= 16 {
                        w.write_all(&data[off..])?;
                        *len -= (data.len() - off) as u64;
                        if *len == 0 {
                            if let Some(sender) = sender.take() {
                                sender.send(*conflicts).unwrap_or(())
                            }
                        }
                    }
                }
                State::None => {
                    debug!("None state");
                }
            }
            Ok((self, session))
        })
    }
}
fn learn(addr: &str, port: u16, pk: &thrussh_keys::key::PublicKey) -> Result<bool, anyhow::Error> {
    if port == 22 {
        print!(
            "Unknown key for {:?}, fingerprint {:?}. Learn it (y/N)? ",
            addr,
            pk.fingerprint()
        );
    } else {
        print!(
            "Unknown key for {:?}:{}, fingerprint {:?}. Learn it (y/N)? ",
            addr,
            port,
            pk.fingerprint()
        );
    }
    std::io::stdout().flush()?;
    let mut buffer = String::new();
    std::io::stdin().read_line(&mut buffer)?;
    let buffer = buffer.trim();
    if buffer == "Y" || buffer == "y" {
        thrussh_keys::learn_known_hosts(addr, port, pk)?;
        Ok(true)
    } else {
        Ok(false)
    }
}
impl Ssh {
    pub async fn finish(&mut self) -> Result<(), anyhow::Error> {
        self.c.eof().await?;
        while let Some(msg) = self.c.wait().await {
            debug!("msg = {:?}", msg);
            match msg {
                thrussh::ChannelMsg::WindowAdjusted { .. } => {}
                thrussh::ChannelMsg::Eof => {}
                thrussh::ChannelMsg::ExitStatus { exit_status } => {
                    if exit_status != 0 {
                        bail!("Remote exited with status {:?}", exit_status)
                    }
                }
                msg => error!("wrong message {:?}", msg),
            }
        }
        Ok(())
    }
    pub async fn get_state(
        &mut self,
        mid: Option<u64>,
    ) -> Result<Option<(u64, Merkle)>, anyhow::Error> {
        debug!("get_state");
        let (sender, receiver) = tokio::sync::oneshot::channel();
        *self.state.lock().await = State::State {
            sender: Some(sender),
        };
        self.run_protocol().await?;
        if let Some(mid) = mid {
            self.c
                .data(format!("state {} {}\n", self.channel, mid).as_bytes())
                .await?;
        } else {
            self.c
                .data(format!("state {}\n", self.channel).as_bytes())
                .await?;
        }
        Ok(receiver.await?)
    }
    pub async fn archive<W: std::io::Write + Send + 'static>(
        &mut self,
        prefix: Option<String>,
        state: Option<(Merkle, &[Hash])>,
        w: W,
    ) -> Result<u64, anyhow::Error> {
        debug!("archive");
        let (sender, receiver) = tokio::sync::oneshot::channel();
        *self.state.lock().await = State::Archive {
            sender: Some(sender),
            len: 0,
            conflicts: 0,
            len_n: 0,
            w: Box::new(w),
        };
        self.run_protocol().await?;
        if let Some((ref state, ref extra)) = state {
            let mut cmd = format!("archive {} {}", self.channel, state.to_base32(),);
            for e in extra.iter() {
                cmd.push_str(&format!(" {}", e.to_base32()));
            }
            if let Some(ref p) = prefix {
                cmd.push_str(" :");
                cmd.push_str(p)
            }
            cmd.push('\n');
            self.c.data(cmd.as_bytes()).await?;
        } else {
            self.c
                .data(
                    format!(
                        "archive {}{}{}\n",
                        self.channel,
                        if prefix.is_some() { " :" } else { "" },
                        prefix.unwrap_or_else(String::new)
                    )
                    .as_bytes(),
                )
                .await?;
        }
        let conflicts = receiver.await.unwrap_or(0);
        Ok(conflicts)
    }
    pub async fn run_protocol(&mut self) -> Result<(), anyhow::Error> {
        if !self.is_running {
            self.is_running = true;
            debug!("run_protocol");
            self.c
                .exec(
                    true,
                    format!(
                        "{} protocol --version {} --repository {}",
                        self.remote_cmd,
                        crate::PROTOCOL_VERSION,
                        self.path
                    ),
                )
                .await?;
            while let Some(msg) = self.c.wait().await {
                debug!("msg = {:?}", msg);
                match msg {
                    thrussh::ChannelMsg::Success => break,
                    thrussh::ChannelMsg::WindowAdjusted { .. } => {}
                    thrussh::ChannelMsg::Eof => {}
                    thrussh::ChannelMsg::ExitStatus { exit_status } => {
                        if exit_status != 0 {
                            bail!("Remote exited with status {:?}", exit_status)
                        }
                    }
                    _ => {}
                }
            }
            debug!("run_protocol done");
        }
        Ok(())
    }
    pub async fn download_changelist<T: MutTxnT>(
        &mut self,
        txn: &mut T,
        remote: &mut RemoteRef<T>,
        from: u64,
        paths: &[String],
    ) -> Result<HashSet<Position<Hash>>, anyhow::Error> {
        let (sender, mut receiver) = tokio::sync::mpsc::channel(10);
        *self.state.lock().await = State::Changelist { sender };
        self.run_protocol().await?;
        debug!("download_changelist");
        let mut command = Vec::new();
        write!(command, "changelist {} {}", self.channel, from).unwrap();
        for p in paths {
            write!(command, " {:?}", p).unwrap()
        }
        command.push(b'\n');
        self.c.data(&command[..]).await?;
        debug!("waiting ssh, command: {:?}", std::str::from_utf8(&command));
        let mut result = HashSet::new();
        while let Some(Some(m)) = receiver.recv().await {
            match m {
                super::ListLine::Change { n, h, m } => {
                    txn.put_remote(remote, n, (h, m))?;
                }
                super::ListLine::Position(pos) => {
                    result.insert(pos);
                }
                super::ListLine::Error(err) => {
                    bail!(err)
                }
            }
        }
        if *self.has_errors.lock().await {
            bail!("Remote sent an error")
        }
        debug!("no msg, result = {:?}", result);
        Ok(result)
    }
    pub async fn upload_changes(
        &mut self,
        mut local: PathBuf,
        to_channel: Option<&str>,
        changes: &[Hash],
    ) -> Result<(), anyhow::Error> {
        self.run_protocol().await?;
        debug!("upload_changes");
        let progress = ProgressBar::new(changes.len() as u64);
        progress.set_style(
            indicatif::ProgressStyle::default_bar()
                .template("  Uploading changes   {wide_bar} {pos:>5}/{len}"),
        );
        let progress = super::PROGRESS.add(progress);
        *PROGRESS.lock().await = Some(progress);
        for c in changes {
            debug!("{:?}", c);
            libpijul::changestore::filesystem::push_filename(&mut local, &c);
            let mut change_file = std::fs::File::open(&local)?;
            let change_len = change_file.metadata()?.len();
            let mut change = thrussh::CryptoVec::new_zeroed(change_len as usize);
            use std::io::Read;
            change_file.read_exact(&mut change[..])?;
            let to_channel = if let Some(t) = to_channel {
                t
            } else {
                self.channel.as_str()
            };
            self.c
                .data(format!("apply {} {} {}\n", to_channel, c.to_base32(), change_len).as_bytes())
                .await?;
            self.c.data(&change[..]).await?;
            if let Some(ref mut progress) = *PROGRESS.lock().await {
                progress.inc(1);
            }
            libpijul::changestore::filesystem::pop_filename(&mut local);
        }
        if let Some(ref mut progress) = *PROGRESS.lock().await {
            progress.set_style(
                indicatif::ProgressStyle::default_bar()
                    .template("✓ Uploading changes {pos:>5}/{len}"),
            );
            progress.finish();
        }
        Ok(())
    }
    pub async fn download_changes(
        &mut self,
        c: &[libpijul::pristine::Hash],
        sender: &mut tokio::sync::mpsc::Sender<libpijul::pristine::Hash>,
        changes_dir: &mut PathBuf,
        full: bool,
    ) -> Result<(), anyhow::Error> {
        self.download_changes_(c, Some(sender), changes_dir, full)
            .await
    }
    async fn download_changes_(
        &mut self,
        c: &[libpijul::pristine::Hash],
        mut sender: Option<&mut tokio::sync::mpsc::Sender<libpijul::pristine::Hash>>,
        changes_dir: &mut PathBuf,
        full: bool,
    ) -> Result<(), anyhow::Error> {
        if c.is_empty() {
            return Ok(());
        }
        let (sender_, mut recv) = tokio::sync::mpsc::channel(100);
        let mut path = changes_dir.clone();
        libpijul::changestore::filesystem::push_filename(&mut path, &c[0]);
        std::fs::create_dir_all(&path.parent().unwrap())?;
        path.set_extension("tmp");
        let file = std::fs::File::create(&path)?;
        *self.state.lock().await = State::Changes {
            sender: Some(sender_),
            remaining_len: 0,
            path,
            file,
            hashes: c.to_vec(),
            current: 0,
        };
        self.run_protocol().await?;
        for c in c {
            debug!("download_change {:?} {:?}", c, full);
            if full {
                self.c
                    .data(format!("change {}\n", c.to_base32()).as_bytes())
                    .await?;
            } else {
                self.c
                    .data(format!("partial {}\n", c.to_base32()).as_bytes())
                    .await?;
            }
        }
        let progress = ProgressBar::new(c.len() as u64);
        progress.set_style(
            indicatif::ProgressStyle::default_bar()
                .template("  Downloading changes {wide_bar} {pos:>5}/{len}"),
        );
        *PROGRESS.lock().await = Some(progress);
        while let Some(_hash) = recv.recv().await {
            debug!("received hash {:?}", _hash);
            if let Some(ref mut progress) = *PROGRESS.lock().await {
                progress.inc(1);
            }
            if let Some(ref mut sender) = sender {
                if sender.send(_hash).await.is_err() {
                    if let Some(ref mut progress) = *PROGRESS.lock().await {
                        progress.abandon();
                    }
                    break;
                }
            }
        }
        if let Some(ref mut progress) = *PROGRESS.lock().await {
            if !progress.is_finished() {
                progress.set_style(
                    indicatif::ProgressStyle::default_bar()
                        .template("✓ Downloading changes {pos:>5}/{len}"),
                );
                progress.finish();
            }
        }
        debug!("done downloading");
        Ok(())
    }
    pub async fn clone_channel<T: TxnTExt + MutTxnTExt>(
        &mut self,
        repo: &mut Repository,
        txn: &mut T,
        channel: &mut ChannelRef<T>,
        lazy: bool,
    ) -> Result<(), anyhow::Error> {
        let (sender, mut recv) = tokio::sync::mpsc::channel(10);
        *self.state.lock().await = State::Channel { sender };
        self.run_protocol().await?;
        debug!("clone channel");
        self.c
            .data(format!("channel {}\n", self.channel).as_bytes())
            .await?;
        let progress = indicatif::ProgressBar::new_spinner();
        progress.set_style(
            indicatif::ProgressStyle::default_spinner().template("{spinner} Cloning channel"),
        );
        progress.enable_steady_tick(100);
        *PROGRESS.lock().await = Some(progress);
        let from_dump_alive = {
            let mut from_dump =
                libpijul::pristine::channel_dump::ChannelFromDump::new(txn, channel.clone());
            debug!("receiving channel");
            while let Some(msg) = recv.recv().await {
                debug!("msg = {:?}", msg.len());
                if from_dump.read(&msg)? {
                    debug!("break");
                    break;
                }
            }
            debug!("from dump done");
            from_dump.alive
        };
        if let Some(ref mut progress) = *PROGRESS.lock().await {
            progress.set_style(
                indicatif::ProgressStyle::default_spinner().template("✓ Cloning channel"),
            );
            progress.finish();
        }
        let channel_ = channel.borrow();
        debug!("cloned, now downloading changes");
        let mut hashes = Vec::new();
        if lazy {
            for &ch in from_dump_alive.iter() {
                let h = txn.get_external(ch)?.unwrap();
                hashes.push(h);
            }
        } else {
            for h in txn.log(&channel_, 0)? {
                hashes.push((h?.1).0);
            }
        }
        std::mem::drop(channel_);
        debug!("hashes = {:#?}", hashes);
        self.download_changes_(&hashes, None, &mut repo.changes_dir, true)
            .await?;
        txn.output_repository_no_pending(
            &mut repo.working_copy,
            &repo.changes,
            channel,
            &mut HashMap::new(),
            "",
            true,
            None,
        )?;
        Ok(())
    }
}

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::bail;
use lazy_static::lazy_static;
use libpijul::pristine::{Base32, ChannelRef, Hash, Merkle, MutTxnT, RemoteRef, TxnT};
use libpijul::DOT_DIR;
use libpijul::{MutTxnTExt, TxnTExt};
use log::{debug, info};
use tokio::sync::Mutex;
use crate::repository::*;
pub mod ssh;
use ssh::*;
pub mod local;
use local::*;
pub mod http;
use http::*;
pub enum RemoteRepo {
    Local(Local),
    Ssh(Ssh),
    Http(Http),
    LocalChannel(String),
    None,
}
lazy_static! {
    static ref PROGRESS: Arc<indicatif::MultiProgress> = Arc::new(indicatif::MultiProgress::new());
    static ref SPINNER: Arc<Mutex<Option<indicatif::ProgressBar>>> = Arc::new(Mutex::new(None));
}
impl Repository {
    pub async fn remote(
        &self,
        self_path: Option<&Path>,
        name: &str,
        channel: &str,
        no_cert_check: bool,
    ) -> Result<RemoteRepo, anyhow::Error> {
        if let Some(name) = self.config.remotes.get(name) {
            unknown_remote(self_path, name, channel, no_cert_check).await
        } else {
            unknown_remote(self_path, name, channel, no_cert_check).await
        }
    }
}
pub async fn unknown_remote(
    self_path: Option<&Path>,
    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);
        return 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(),
        }));
    }
    if name.starts_with("ssh://") {
        return if let Some(mut ssh) = ssh_remote(name) {
            debug!("unknown_remote, ssh = {:?}", ssh);
            Ok(RemoteRepo::Ssh(ssh.connect(name, channel).await?))
        } else {
            bail!("Remote not found: {:?}", name)
        };
    }
    if let Ok(root) = std::fs::canonicalize(name) {
        if let Some(path) = self_path {
            let path = std::fs::canonicalize(path)?;
            if path == root {
                return Ok(RemoteRepo::LocalChannel(channel.to_string()));
            }
        }
        let mut dot_dir = root.join(DOT_DIR);
        let changes_dir = dot_dir.join(CHANGES_DIR);
        dot_dir.push(PRISTINE_DIR);
        debug!("dot_dir = {:?}", dot_dir);
        if let Ok(pristine) = libpijul::pristine::sanakirja::Pristine::new(&dot_dir) {
            debug!("pristine done");
            return Ok(RemoteRepo::Local(Local {
                root: Path::new(name).to_path_buf(),
                channel: channel.to_string(),
                changes_dir,
                pristine: Arc::new(pristine),
                name: name.to_string(),
            }));
        }
    }
    if let Some(mut ssh) = ssh_remote(name) {
        debug!("unknown_remote, ssh = {:?}", ssh);
        Ok(RemoteRepo::Ssh(ssh.connect(name, channel).await?))
    } else {
        bail!("Remote not found: {:?}", name)
    }
}
impl RemoteRepo {
    fn name(&self) -> Option<&str> {
        match *self {
            RemoteRepo::Ssh(ref s) => Some(s.name.as_str()),
            RemoteRepo::Local(ref l) => Some(l.name.as_str()),
            RemoteRepo::Http(ref h) => Some(h.name.as_str()),
            RemoteRepo::LocalChannel(_) => None,
            RemoteRepo::None => unreachable!(),
        }
    }
    pub fn repo_name(&self) -> Option<String> {
        match *self {
            RemoteRepo::Ssh(ref s) => {
                if let Some(sep) = s.name.rfind(|c| c == ':' || c == '/') {
                    Some(s.name.split_at(sep + 1).1.to_string())
                } else {
                    Some(s.name.as_str().to_string())
                }
            }
            RemoteRepo::Local(ref l) => {
                if let Some(file) = l.root.file_name() {
                    Some(file.to_str().unwrap().to_string())
                } else {
                    None
                }
            }
            RemoteRepo::Http(ref h) => {
                let url = reqwest::Url::parse(&h.url).unwrap();
                if let Some(name) = libpijul::path::file_name(url.path()) {
                    Some(name.to_string())
                } else {
                    url.host().map(|h| h.to_string())
                }
            }
            RemoteRepo::LocalChannel(_) => None,
            RemoteRepo::None => unreachable!(),
        }
    }
    pub async fn finish(&mut self) -> Result<(), anyhow::Error> {
        if let RemoteRepo::Ssh(s) = self {
            s.finish().await?
        }
        Ok(())
    }
    pub async fn update_changelist<T: MutTxnTExt + TxnTExt>(
        &mut self,
        txn: &mut T,
        path: &[String],
    ) -> Result<Option<(HashSet<Position<Hash>>, RemoteRef<T>)>, anyhow::Error> {
        let progress = indicatif::ProgressBar::new_spinner();
        progress.set_style(
            indicatif::ProgressStyle::default_spinner()
                .template("{spinner} Updating remote changelist"),
        );
        progress.enable_steady_tick(100);
        *SPINNER.lock().await = Some(progress);
        debug!("update_changelist");
        let name = if let Some(name) = self.name() {
            name
        } else {
            return Ok(None);
        };
        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| {
                let k = k.unwrap().0;
                if k >= n {
                    Some(k)
                } else {
                    None
                }
            })
            .collect();
        for k in v {
            debug!("deleting {:?}", k);
            txn.del_remote(&mut remote, k)?;
        }
        let paths = self.download_changelist(txn, &mut remote, n, path).await?;
        if let Some(progress) = SPINNER.lock().await.take() {
            progress.set_style(
                indicatif::ProgressStyle::default_spinner()
                    .template("✓ Updating remote changelist"),
            );
            progress.finish();
        }
        Ok(Some((paths, remote)))
    }
    async fn dichotomy_changelist<T: MutTxnT + TxnTExt>(
        &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(txn, 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(txn, 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<T: libpijul::TxnTExt>(
        &mut self,
        txn: &T,
        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 mut h) => h.get_state(mid).await,
            RemoteRepo::LocalChannel(ref channel) => {
                if let Some(channel) = txn.load_channel(&channel)? {
                    local::get_state(txn, &channel, mid)
                } else {
                    Ok(None)
                }
            }
            RemoteRepo::None => unreachable!(),
        }
    }
    pub async fn archive<W: std::io::Write + Send + 'static>(
        &mut self,
        prefix: Option<String>,
        state: Option<(Merkle, &[Hash])>,
        w: W,
    ) -> Result<u64, anyhow::Error> {
        match *self {
            RemoteRepo::Local(ref mut l) => {
                debug!("archiving local repo");
                let changes = libpijul::changestore::filesystem::FileSystem::from_root(&l.root);
                let mut tarball = libpijul::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 mut h) => h.archive(prefix, state, w).await,
            RemoteRepo::LocalChannel(_) => unreachable!(),
            RemoteRepo::None => unreachable!(),
        }
    }
    async fn download_changelist<T: MutTxnTExt>(
        &mut self,
        txn: &mut T,
        remote: &mut RemoteRef<T>,
        from: u64,
        paths: &[String],
    ) -> Result<HashSet<Position<Hash>>, 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) => h.download_changelist(txn, remote, from, paths).await,
            RemoteRepo::LocalChannel(_) => Ok(HashSet::new()),
            RemoteRepo::None => unreachable!(),
        }
    }
    pub async fn upload_changes<T: MutTxnTExt>(
        &mut self,
        txn: &mut T,
        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) => h.upload_changes(local, to_channel, changes).await,
            RemoteRepo::LocalChannel(ref channel) => {
                let mut channel = txn.open_or_create_channel(channel)?;
                let store = libpijul::changestore::filesystem::FileSystem::from_changes(local);
                local::upload_changes(&store, txn, &mut channel, changes)
            }
            RemoteRepo::None => unreachable!(),
        }
    }
    /// Start (and possibly complete) the download of a change.
    pub async fn download_changes(
        &mut self,
        hashes: &[libpijul::pristine::Hash],
        send: &mut tokio::sync::mpsc::Sender<libpijul::pristine::Hash>,
        path: &mut PathBuf,
        full: bool,
    ) -> Result<bool, anyhow::Error> {
        debug!("download_changes");
        match *self {
            RemoteRepo::Local(ref mut l) => l.download_changes(hashes, send, path).await?,
            RemoteRepo::Ssh(ref mut s) => s.download_changes(hashes, send, path, full).await?,
            RemoteRepo::Http(ref mut h) => h.download_changes(hashes, send, path, full).await?,
            RemoteRepo::LocalChannel(_) => {}
            RemoteRepo::None => unreachable!(),
        }
        libpijul::changestore::filesystem::pop_filename(path);
        Ok(true)
    }
    pub async fn pull<T: MutTxnTExt + TxnTExt>(
        &mut self,
        repo: &mut Repository,
        txn: &mut T,
        channel: &mut ChannelRef<T>,
        to_apply: &[Hash],
        inodes: &HashSet<Position<Hash>>,
        do_apply: bool,
    ) -> Result<Vec<Hash>, anyhow::Error> {
        let (mut send, mut recv) = tokio::sync::mpsc::channel(100);
        let mut change_path_ = repo.changes_dir.clone();
        let mut to_download_ = Vec::new();
        for h in to_apply {
            libpijul::changestore::filesystem::push_filename(&mut change_path_, h);
            if std::fs::metadata(&change_path_).is_err() {
                to_download_.push(*h)
            }
            libpijul::changestore::filesystem::pop_filename(&mut change_path_);
        }
        let to_download = to_download_.clone();
        let mut self_ = std::mem::replace(self, RemoteRepo::None);
        let t = tokio::spawn(async move {
            self_
                .download_changes(&to_download_, &mut send, &mut change_path_, false)
                .await?;
            Ok::<_, anyhow::Error>(self_)
        });
        let mut ws = libpijul::ApplyWorkspace::new();
        let mut change_path = repo.changes_dir.clone();
        let progress = if do_apply {
            let p = indicatif::ProgressBar::new(to_download.len() as u64);
            p.set_style(
                indicatif::ProgressStyle::default_bar()
                    .template("  Applying changes    {wide_bar} {pos:>5}/{len}"),
            );
            Some(PROGRESS.add(p))
        } else {
            None
        };
        let t_progress = std::thread::spawn(|| {
            PROGRESS.join().unwrap_or(());
        });
        let mut to_apply_inodes = Vec::new();
        for h in to_apply {
            libpijul::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() {
                    if let Some(ref progress) = progress {
                        progress.abandon();
                    }
                    break;
                }
            }
            libpijul::changestore::filesystem::pop_filename(&mut change_path);
            let touches_inodes = inodes.is_empty()
                || {
                    debug!("inodes = {:?}", inodes);
                    use libpijul::changestore::ChangeStore;
                    let changes = repo.changes.get_changes(h)?;
                    changes.iter().any(|c| {
                        c.iter().any(|c| {
                            let inode = c.inode();
                            debug!("inode = {:?}", inode);
                            if let Some(h) = inode.change {
                                inodes.contains(&Position {
                                    change: h,
                                    pos: inode.pos,
                                })
                            } else {
                                false
                            }
                        })
                    })
                }
                || { inodes.iter().any(|i| i.change == *h) };
            if touches_inodes {
                to_apply_inodes.push(*h);
            } else {
                continue;
            }
            if let Some(ref progress) = progress {
                info!("Applying {:?}", h);
                progress.inc(1);
                txn.apply_change_ws(&repo.changes, channel, *h, &mut ws)?;
            } else {
                debug!("not applying {:?}", h)
            }
        }
        if let Some(progress) = progress {
            if !progress.is_finished() {
                progress.set_style(
                    indicatif::ProgressStyle::default_bar()
                        .template("✓ Applying changes    {pos:>5}/{len}"),
                );
                progress.finish()
            }
        }
        std::mem::drop(recv);
        debug!("waiting for spawned process");
        *self = t.await??;
        t_progress.join().unwrap();
        Ok(to_apply_inodes)
    }
    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 (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 {
            while let Some(hash) = recv_hash.recv().await {
                self_
                    .download_changes(&[hash], &mut send_signal, &mut change_path_, false)
                    .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 {
            libpijul::changestore::filesystem::push_filename(&mut change_path, &hash);
            std::fs::create_dir_all(change_path.parent().unwrap())?;
            use libpijul::changestore::ChangeStore;
            hashes.push(hash);
            for dep in repo.changes.get_dependencies(&hash)? {
                let dep: libpijul::pristine::Hash = dep;
                send_hash.send(dep).await?;
            }
            libpijul::changestore::filesystem::pop_filename(&mut change_path);
        }
        std::mem::drop(recv_signal);
        std::mem::drop(send_hash);
        let mut ws = libpijul::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().unwrap();
        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 x in txn.iter_rev_remote(&remote.borrow().remote, None)? {
                let (n, (h, s)) = x?;
                debug!("{:?} {:?} {:?}", n, h, s);
                if s == state {
                    found = true;
                    break;
                }
                to_unrecord.push(h);
            }
            if !found {
                bail!("State not found: {:?}", state)
            }
            self.pull(repo, txn, channel, &to_unrecord, &HashSet::new(), 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 x in txn.iter_remote(&remote.borrow().remote, 0)? {
            let (n, (h, s)) = x?;
            debug!("{:?} {:?} {:?}", n, h, s);
            to_pull.push(h);
            if s == state {
                found = true;
                break;
            }
        }
        if !found {
            bail!("State not found: {:?}", state)
        }
        self.pull(repo, txn, channel, &to_pull, &HashSet::new(), true)
            .await?;
        Ok(())
    }
    pub async fn complete_changes<T: MutTxnT + TxnTExt>(
        &mut self,
        repo: &crate::repository::Repository,
        txn: &T,
        local_channel: &mut ChannelRef<T>,
        changes: &[Hash],
        full: bool,
    ) -> Result<(), anyhow::Error> {
        use libpijul::changestore::ChangeStore;
        let (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 {
            while let Some(h) = recv_hash.recv().await {
                debug!("downloading full patch: {:?}", h);
                self_
                    .download_changes(&[h], &mut send_sig, &mut changes_dir, true)
                    .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 = libpijul::pristine::Vertex {
                change,
                start: libpijul::pristine::ChangePosition(0),
                end: libpijul::pristine::ChangePosition(0),
            };
            let channel = local_channel.borrow();
            if txn.is_alive(&channel, v)? {
                send_hash.send(*c).await?;
            }
        }
        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() {
            if let RemoteRepo::Ssh(s) = self {
                return s.clone_channel(repo, txn, local_channel, lazy).await;
            }
        }
        let (inodes, remote_changes) = self
            .update_changelist(txn, path)
            .await?
            .expect("Remote is not self");
        let mut pullable = Vec::new();
        for x in txn.iter_remote(&remote_changes.borrow().remote, 0)? {
            let (_, (h, _)) = x?;
            pullable.push(h)
        }
        self.pull(repo, txn, local_channel, &pullable, &inodes, true)
            .await?;
        Ok(())
    }
}
use libpijul::pristine::{ChangePosition, Position};
use regex::Regex;
lazy_static! {
    static ref CHANGELIST_LINE: Regex =
        Regex::new(r#"(?P<num>[0-9]+)\.(?P<hash>[A-Za-z0-9]+)\.(?P<merkle>[A-Za-z0-9]+)"#).unwrap();
    static ref PATHS_LINE: Regex =
        Regex::new(r#"(?P<hash>[A-Za-z0-9]+)\.(?P<num>[0-9]+)"#).unwrap();
}
enum ListLine {
    Change { n: u64, h: Hash, m: Merkle },
    Position(Position<Hash>),
    Error(String),
}
fn parse_line(data: &str) -> Result<ListLine, anyhow::Error> {
    debug!("data = {:?}", data);
    if let Some(caps) = CHANGELIST_LINE.captures(data) {
        if let (Some(h), Some(m)) = (
            Hash::from_base32(caps.name("hash").unwrap().as_str().as_bytes()),
            Merkle::from_base32(caps.name("merkle").unwrap().as_str().as_bytes()),
        ) {
            return Ok(ListLine::Change {
                n: caps.name("num").unwrap().as_str().parse().unwrap(),
                h,
                m,
            });
        }
    }
    if data.starts_with("error:") {
        return Ok(ListLine::Error(data.split_at(6).1.to_string()));
    }
    if let Some(caps) = PATHS_LINE.captures(data) {
        return Ok(ListLine::Position(Position {
            change: Hash::from_base32(caps.name("hash").unwrap().as_str().as_bytes()).unwrap(),
            pos: ChangePosition(caps.name("num").unwrap().as_str().parse().unwrap()),
        }));
    }
    debug!("offending line: {:?}", data);
    bail!("Protocol error")
}

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::Arc;
use libpijul::pristine::{Hash, Merkle, MutTxnT, Position, TxnT};
use libpijul::*;
use log::debug;
use super::RemoteRef;
#[derive(Clone)]
pub struct Local {
    pub channel: String,
    pub root: std::path::PathBuf,
    pub changes_dir: std::path::PathBuf,
    pub pristine: Arc<libpijul::pristine::sanakirja::Pristine>,
    pub name: String,
}
pub fn get_state<T: TxnTExt>(
    txn: &T,
    channel: &libpijul::pristine::ChannelRef<T>,
    mid: Option<u64>,
) -> Result<Option<(u64, Merkle)>, anyhow::Error> {
    if let Some(mid) = mid {
        Ok(txn.get_changes(&channel, mid)?.map(|(_, m)| (mid, m)))
    } else {
        Ok(txn.reverse_log(&channel.borrow(), None)?.next().map(|n| {
            let (n, (_, m)) = n.unwrap();
            (n, m)
        }))
    }
}
impl Local {
    pub fn get_state(&mut self, mid: Option<u64>) -> Result<Option<(u64, Merkle)>, anyhow::Error> {
        let txn = self.pristine.txn_begin()?;
        let channel = txn.load_channel(&self.channel)?.unwrap();
        Ok(get_state(&txn, &channel, mid)?)
    }
    pub fn download_changelist<T: MutTxnT>(
        &mut self,
        txn: &mut T,
        remote: &mut RemoteRef<T>,
        from: u64,
        paths: &[String],
    ) -> Result<HashSet<Position<Hash>>, anyhow::Error> {
        let store = libpijul::changestore::filesystem::FileSystem::from_root(&self.root);
        let remote_txn = self.pristine.txn_begin()?;
        let remote_channel = if let Some(channel) = remote_txn.load_channel(&self.channel)? {
            channel
        } else {
            debug!("no remote channel named {:?}", self.channel);
            return Ok(HashSet::new());
        };
        let mut paths_ = HashSet::new();
        let mut result = HashSet::new();
        for s in paths {
            if let Ok((p, _ambiguous)) = remote_txn.follow_oldest_path(&store, &remote_channel, s) {
                debug!("p = {:?}", p);
                result.insert(Position {
                    change: remote_txn.get_external(p.change)?.unwrap(),
                    pos: p.pos,
                });
                paths_.insert(p);
                paths_.extend(
                    libpijul::fs::iter_graph_descendants(
                        &remote_txn,
                        &remote_channel.borrow().graph,
                        p,
                    )?
                    .map(|x| x.unwrap()),
                );
            }
        }
        debug!("paths_ = {:?}", paths_);
        for x in remote_txn.log(&remote_channel.borrow(), from)? {
            let (n, (h, m)) = x?;
            if n >= from {
                debug!(
                    "downloading changelist item {:?} {:?} {:?}",
                    n,
                    h.to_base32(),
                    m.to_base32()
                );
                let h_int = remote_txn.get_internal(h)?.unwrap();
                if paths_.is_empty()
                    || paths_.iter().any(|x| {
                        remote_txn
                            .get_touched_files(*x, Some(h_int))
                            .unwrap()
                            .is_some()
                    })
                {
                    txn.put_remote(remote, n, (h, m))?;
                }
            }
        }
        Ok(result)
    }
    pub fn upload_changes(
        &mut self,
        mut local: PathBuf,
        to_channel: Option<&str>,
        changes: &[Hash],
    ) -> Result<(), anyhow::Error> {
        let store = libpijul::changestore::filesystem::FileSystem::from_root(&self.root);
        let mut txn = self.pristine.mut_txn_begin();
        let mut channel = txn.open_or_create_channel(to_channel.unwrap_or(&self.channel))?;
        for c in changes {
            libpijul::changestore::filesystem::push_filename(&mut local, &c);
            libpijul::changestore::filesystem::push_filename(&mut self.changes_dir, &c);
            std::fs::create_dir_all(&self.changes_dir.parent().unwrap())?;
            debug!("hard link {:?} {:?}", local, self.changes_dir);
            if std::fs::metadata(&self.changes_dir).is_err() {
                if std::fs::hard_link(&local, &self.changes_dir).is_err() {
                    std::fs::copy(&local, &self.changes_dir)?;
                }
            }
            debug!("hard link done");
            libpijul::changestore::filesystem::pop_filename(&mut local);
            libpijul::changestore::filesystem::pop_filename(&mut self.changes_dir);
        }
        let mut repo = libpijul::working_copy::filesystem::FileSystem::from_root(&self.root);
        upload_changes(&store, &mut txn, &mut channel, changes)?;
        txn.output_repository_no_pending(
            &mut repo,
            &store,
            &mut channel,
            &mut HashMap::new(),
            "",
            true,
            None,
        )?;
        txn.commit()?;
        Ok(())
    }
    pub async fn download_changes(
        &mut self,
        c: &[libpijul::pristine::Hash],
        send: &mut tokio::sync::mpsc::Sender<libpijul::pristine::Hash>,
        mut path: &mut PathBuf,
    ) -> Result<(), anyhow::Error> {
        for c in c {
            libpijul::changestore::filesystem::push_filename(&mut self.changes_dir, c);
            libpijul::changestore::filesystem::push_filename(&mut path, c);
            if std::fs::metadata(&path).is_ok() {
                debug!("metadata {:?} ok", path);
                libpijul::changestore::filesystem::pop_filename(&mut path);
                continue;
            }
            std::fs::create_dir_all(&path.parent().unwrap())?;
            if std::fs::hard_link(&self.changes_dir, &path).is_err() {
                std::fs::copy(&self.changes_dir, &path)?;
            }
            debug!("hard link done");
            libpijul::changestore::filesystem::pop_filename(&mut self.changes_dir);
            libpijul::changestore::filesystem::pop_filename(&mut path);
            debug!("sent");
            send.send(*c).await.unwrap();
        }
        Ok(())
    }
}
pub fn upload_changes<T: MutTxnTExt, C: libpijul::changestore::ChangeStore>(
    store: &C,
    txn: &mut T,
    channel: &mut libpijul::pristine::ChannelRef<T>,
    changes: &[Hash],
) -> Result<(), anyhow::Error> {
    let mut ws = libpijul::ApplyWorkspace::new();
    for c in changes {
        txn.apply_change_ws(store, channel, *c, &mut ws)?;
    }
    Ok(())
}

use std::collections::HashSet;
use std::io::Write;
use std::path::PathBuf;
use anyhow::bail;
use indicatif::ProgressBar;
use libpijul::pristine::{Base32, MutTxnT, Position};
use libpijul::{Hash, RemoteRef};
use log::{debug, error};
pub struct Http {
    pub url: String,
    pub channel: String,
    pub client: reqwest::Client,
    pub name: String,
}
async fn download_change(
    client: reqwest::Client,
    url: String,
    mut path: PathBuf,
    c: libpijul::pristine::Hash,
) -> Result<libpijul::pristine::Hash, anyhow::Error> {
    libpijul::changestore::filesystem::push_filename(&mut path, &c);
    std::fs::create_dir_all(&path.parent().unwrap())?;
    let path_ = path.with_extension("tmp");
    let mut f = std::fs::File::create(&path_)?;
    libpijul::changestore::filesystem::pop_filename(&mut path);
    let c32 = c.to_base32();
    let url = format!("{}/{}", url, super::DOT_DIR);
    let mut delay = 1f64;
    loop {
        let mut res = if let Ok(res) = client.get(&url).query(&[("change", &c32)]).send().await {
            delay = 1f64;
            res
        } else {
            error!("HTTP error, retrying in {} seconds", delay.round());
            tokio::time::sleep(std::time::Duration::from_secs_f64(delay)).await;
            delay *= 2.;
            continue;
        };
        debug!("response {:?}", res);
        if !res.status().is_success() {
            bail!("HTTP error {:?}", res.status())
        }
        let done = loop {
            match res.chunk().await {
                Ok(Some(chunk)) => {
                    debug!("writing {:?}", chunk.len());
                    f.write_all(&chunk)?;
                }
                Ok(None) => break true,
                Err(_) => {
                    error!("Error while downloading {:?}, retrying", url);
                    tokio::time::sleep(std::time::Duration::from_secs_f64(delay)).await;
                    delay *= 2.;
                    break false;
                }
            }
        };
        if done {
            std::fs::rename(&path_, &path_.with_extension("change"))?;
            break;
        }
    }
    Ok(c)
}
const POOL_SIZE: usize = 20;
impl Http {
    pub async fn download_changes(
        &mut self,
        hashes: &[libpijul::pristine::Hash],
        send: &mut tokio::sync::mpsc::Sender<libpijul::pristine::Hash>,
        path: &PathBuf,
        _full: bool,
    ) -> Result<(), anyhow::Error> {
        let progress = ProgressBar::new(hashes.len() as u64);
        progress.set_style(
            indicatif::ProgressStyle::default_bar()
                .template("  Downloading changes {wide_bar} {pos:>5}/{len}"),
        );
        let progress = super::PROGRESS.add(progress);
        let mut pool = <[_; POOL_SIZE]>::default();
        let mut cur = 0;
        for c in hashes {
            debug!("downloading {:?}", c);
            progress.inc(1);
            let t = std::mem::replace(
                &mut pool[cur],
                Some(tokio::spawn(download_change(
                    self.client.clone(),
                    self.url.clone(),
                    path.clone(),
                    *c,
                ))),
            );
            if let Some(t) = t {
                let c = t.await??;
                if send.send(c).await.is_err() {
                    debug!("err for {:?}", c);
                    progress.abandon();
                    break;
                }
            }
            cur = (cur + 1) % POOL_SIZE;
        }
        for f in 0..POOL_SIZE {
            if let Some(t) = pool[(cur + f) % POOL_SIZE].take() {
                let c = t.await??;
                if send.send(c).await.is_err() {
                    debug!("err for {:?}", c);
                    progress.abandon();
                    break;
                }
            }
        }
        if !progress.is_finished() {
            progress.set_style(
                indicatif::ProgressStyle::default_bar()
                    .template("✓ Downloading changes {pos:>5}/{len}"),
            );
            progress.finish();
        }
        Ok(())
    }
    pub async fn upload_changes(
        &self,
        mut local: PathBuf,
        to_channel: Option<&str>,
        changes: &[libpijul::Hash],
    ) -> Result<(), anyhow::Error> {
        let progress = ProgressBar::new(changes.len() as u64);
        progress.set_style(
            indicatif::ProgressStyle::default_bar()
                .template("  Uploading changes   {wide_bar} {pos:>5}/{len}"),
        );
        for c in changes {
            libpijul::changestore::filesystem::push_filename(&mut local, &c);
            let url = self.url.clone() + "/" + super::DOT_DIR;
            let change = std::fs::read(&local)?;
            let mut to_channel = if let Some(ch) = to_channel {
                vec![("to_channel", ch)]
            } else {
                Vec::new()
            };
            let c = c.to_base32();
            to_channel.push(("apply", &c));
            debug!("url {:?} {:?}", url, to_channel);
            self.client
                .post(&url)
                .query(&to_channel)
                .body(change)
                .send()
                .await?;
            progress.inc(1);
            libpijul::changestore::filesystem::pop_filename(&mut local);
        }
        progress.set_style(
            indicatif::ProgressStyle::default_bar().template("✓ Uploading changes {pos:>5}/{len}"),
        );
        progress.finish();
        Ok(())
    }
    pub async fn download_changelist<T: MutTxnT>(
        &self,
        txn: &mut T,
        remote: &mut RemoteRef<T>,
        from: u64,
        paths: &[String],
    ) -> Result<HashSet<Position<Hash>>, anyhow::Error> {
        let url = self.url.clone() + "/" + super::DOT_DIR;
        let from_ = from.to_string();
        let mut query = vec![("changelist", &from_), ("channel", &self.channel)];
        for p in paths.iter() {
            query.push(("path", p));
        }
        let res = self.client.get(&url).query(&query).send().await?;
        if !res.status().is_success() {
            bail!("HTTP error {:?}", res.status())
        }
        let resp = res.bytes().await?;
        let mut result = HashSet::new();
        if let Ok(data) = std::str::from_utf8(&resp) {
            for l in data.lines() {
                if !l.is_empty() {
                    match super::parse_line(l)? {
                        super::ListLine::Change { n, m, h } => {
                            txn.put_remote(remote, n, (h, m))?;
                        }
                        super::ListLine::Position(pos) => {
                            result.insert(pos);
                        }
                        super::ListLine::Error(e) => {
                            let mut stderr = std::io::stderr();
                            writeln!(stderr, "{}", e)?;
                        }
                    }
                } else {
                    break;
                }
            }
        }
        Ok(result)
    }
    pub async fn get_state(
        &mut self,
        mid: Option<u64>,
    ) -> Result<Option<(u64, libpijul::Merkle)>, anyhow::Error> {
        debug!("get_state {:?}", self.url);
        let url = format!("{}/{}", self.url, super::DOT_DIR);
        let q = if let Some(mid) = mid {
            [
                ("state", format!("{}", mid)),
                ("channel", self.channel.clone()),
            ]
        } else {
            [("state", String::new()), ("channel", self.channel.clone())]
        };
        let res = self.client.get(&url).query(&q).send().await?;
        if !res.status().is_success() {
            bail!("HTTP error {:?}", res.status())
        }
        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| libpijul::Merkle::from_base32(m.as_bytes())),
        ) {
            Ok(Some((n, m)))
        } else {
            Ok(None)
        }
    }
    pub async fn archive<W: std::io::Write + Send + 'static>(
        &mut self,
        prefix: Option<String>,
        state: Option<(libpijul::Merkle, &[Hash])>,
        mut w: W,
    ) -> Result<u64, anyhow::Error> {
        let url = self.url.clone() + "/" + super::DOT_DIR;
        let res = self.client.get(&url).query(&[("channel", &self.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?;
        if !res.status().is_success() {
            bail!("HTTP error {:?}", res.status())
        }
        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)
    }
}

mod commands;
mod config;
mod remote;
mod repository;
use std::ffi::OsString;
use std::io::Write;
use std::path::PathBuf;
use anyhow::bail;
use clap::{AppSettings, Clap};
use human_panic::setup_panic;
use crate::commands::*;
const DEFAULT_CHANNEL: &str = "main";
const PROTOCOL_VERSION: usize = 3;
#[derive(Clap, Debug)]
#[clap(
    version,
    author,
    global_setting(AppSettings::ColoredHelp),
    setting(AppSettings::InferSubcommands)
)]
pub struct Opts {
    #[clap(subcommand)]
    pub subcmd: SubCommand,
}
#[derive(Clap, Debug)]
pub enum SubCommand {
    /// Initializes an empty pijul repository
    Init(Init),
    /// Clones an existing pijul repository
    Clone(Clone),
    /// Creates a new change
    Record(Record),
    /// Shows difference between two channels/changes
    Diff(Diff),
    /// Show the entire log of changes
    Log(Log),
    /// Pushes changes to a remote upstream
    Push(Push),
    /// Pulls changes from a remote upstream
    Pull(Pull),
    /// Shows information about a particular change
    Change(Change),
    /// Manages different channels
    Channel(Channel),
    #[clap(setting = AppSettings::Hidden)]
    Protocol(Protocol),
    #[cfg(feature = "git")]
    /// Imports a git repository into pijul
    Git(Git),
    /// Moves a file in the working copy and the tree
    Mv(Mv),
    /// Lists files tracked by pijul
    Ls(Ls),
    /// Adds a path to the tree.
    ///
    /// Pijul has an internal tree to represent the files currently
    /// tracked. This command adds files and directories to that tree.
    Add(Add),
    /// Removes a file from the tree and pristine
    Remove(Remove),
    /// Resets the working copy to the last recorded change.
    ///
    /// In other words, discards all unrecorded changes.
    Reset(Reset),
    #[cfg(debug_assertions)]
    Debug(Debug),
    /// Create a new channel
    Fork(Fork),
    /// Unrecords a list of changes.
    ///
    /// The changes will be removed from your log, but your working
    /// copy will stay exactly the same, unless the
    /// `--reset` flag was passed. A change can only be unrecorded
    /// if all changes that depend on it are also unrecorded in the
    /// same operation. There are two ways to call `pijul-unrecord`:
    ///
    /// * With a list of <change-id>s. The given changes will be
    /// unrecorded, if possible.
    ///
    /// * Without listing any <change-id>s. You will be
    /// presented with a list of changes to choose from.
    /// The length of the list is determined by the `unrecord_changes`
    /// setting in your global config or the `--show-changes` option,
    /// with the latter taking precedence.
    Unrecord(Unrecord),
    /// Applies changes to a channel
    Apply(Apply),
    /// Manages remote repositories
    Remote(Remote),
    /// Creates an archive of the repository
    Archive(Archive),
    /// Shows which patch last affected each line of the every file
    Credit(Credit),
    #[clap(external_subcommand)]
    ExternalSubcommand(Vec<OsString>),
}
#[tokio::main]
async fn main() {
    setup_panic!();
    env_logger::init();
    let opts = Opts::parse();
    if let Err(e) = run(opts).await {
        writeln!(std::io::stderr(), "Error: {}", e).unwrap_or(());
        std::process::exit(1);
    } else {
        std::process::exit(0);
    }
}
#[cfg(unix)]
fn run_external_command(mut command: Vec<OsString>) -> Result<(), std::io::Error> {
    let args = command.split_off(1);
    let mut cmd: OsString = "pijul-".into();
    cmd.push(&command[0]);
    use std::os::unix::process::CommandExt;
    let err = std::process::Command::new(&cmd).args(args).exec();
    report_external_command_error(&command[0], err);
}
#[cfg(windows)]
fn run_external_command(mut command: Vec<OsString>) -> Result<(), std::io::Error> {
    let args = command.split_off(1);
    let mut cmd: OsString = "pijul-".into();
    cmd.push(&command[0]);
    let mut spawned = match std::process::Command::new(&cmd).args(args).spawn() {
        Ok(spawned) => spawned,
        Err(e) => {
            report_external_command_error(&command[0], e);
        }
    };
    let status = spawned.wait()?;
    std::process::exit(status.code().unwrap_or(1))
}
fn report_external_command_error(cmd: &OsString, err: std::io::Error) -> ! {
    if err.kind() == std::io::ErrorKind::NotFound {
        writeln!(std::io::stderr(), "No such subcommand: {:?}", cmd).unwrap_or(());
    } else {
        writeln!(std::io::stderr(), "Error while running {:?}: {}", cmd, err).unwrap_or(());
    }
    std::process::exit(1)
}
async fn run(opts: Opts) -> Result<(), anyhow::Error> {
    match opts.subcmd {
        SubCommand::Log(l) => l.run(),
        SubCommand::Init(init) => init.run(),
        SubCommand::Clone(clone) => clone.run().await,
        SubCommand::Record(record) => record.run().await,
        SubCommand::Diff(diff) => diff.run(),
        SubCommand::Push(push) => push.run().await,
        SubCommand::Pull(pull) => pull.run().await,
        SubCommand::Change(change) => change.run(),
        SubCommand::Channel(channel) => channel.run(),
        SubCommand::Protocol(protocol) => protocol.run(),
        #[cfg(feature = "git")]
        SubCommand::Git(git) => git.run(),
        SubCommand::Mv(mv) => mv.run(),
        SubCommand::Ls(ls) => ls.run(),
        SubCommand::Add(add) => add.run(),
        SubCommand::Remove(remove) => remove.run(),
        SubCommand::Reset(reset) => reset.run(),
        #[cfg(debug_assertions)]
        SubCommand::Debug(debug) => debug.run(),
        SubCommand::Fork(fork) => fork.run(),
        SubCommand::Unrecord(unrecord) => unrecord.run(),
        SubCommand::Apply(apply) => apply.run(),
        SubCommand::Remote(remote) => remote.run(),
        SubCommand::Archive(archive) => archive.run().await,
        SubCommand::Credit(credit) => credit.run(),
        SubCommand::ExternalSubcommand(command) => Ok(run_external_command(command)?),
    }
}
pub fn current_dir() -> Result<PathBuf, anyhow::Error> {
    if let Ok(cur) = std::env::current_dir() {
        Ok(cur)
    } else {
        bail!("Cannot access working directory")
    }
}

use std::collections::HashMap;
use anyhow::bail;
use log::debug;
use serde_derive::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize)]
pub struct Global {
    pub author: libpijul::change::Author,
    pub unrecord_changes: Option<usize>,
}
const CONFIG_DIR: &str = "pijul";
impl Global {
    pub fn load() -> Result<Global, anyhow::Error> {
        if let Some(mut dir) = dirs_next::config_dir() {
            dir.push(CONFIG_DIR);
            dir.push("config.toml");
            let s = std::fs::read(&dir)
                .or_else(|e| {
                    // Read from `$HOME/.config/pijul` dir
                    if let Some(mut dir) = dirs_next::home_dir() {
                        dir.push(".config");
                        dir.push(CONFIG_DIR);
                        dir.push("config.toml");
                        std::fs::read(&dir)
                    } else {
                        Err(e.into())
                    }
                })
                .or_else(|e| {
                    // Read from `$HOME/.pijulconfig`
                    if let Some(mut dir) = dirs_next::home_dir() {
                        dir.push(".pijulconfig");
                        std::fs::read(&dir)
                    } else {
                        Err(e.into())
                    }
                })?;
            debug!("s = {:?}", s);
            if let Ok(t) = toml::from_slice(&s) {
                Ok(t)
            } else {
                bail!("Could not read configuration file at {:?}", dir)
            }
        } else {
            bail!("Global configuration file missing")
        }
    }
}
#[derive(Debug, Serialize, Deserialize, Default)]
pub struct Config {
    pub current_channel: Option<String>,
    pub default_remote: Option<String>,
    #[serde(default)]
    pub remotes: HashMap<String, String>,
    #[serde(default)]
    pub hooks: Hooks,
}
#[derive(Debug, Serialize, Deserialize, Default)]
pub struct Hooks {
    #[serde(default)]
    pub record: Vec<String>,
}
impl Config {
    pub fn get_current_channel<'a>(&'a self, alt: Option<&'a String>) -> &'a str {
        if let Some(channel) = alt {
            channel.as_ref()
        } else if let Some(ref channel) = self.current_channel {
            channel.as_str()
        } else {
            crate::DEFAULT_CHANNEL
        }
    }
    pub fn current_channel(&self) -> Option<&str> {
        if let Some(ref channel) = self.current_channel {
            Some(channel.as_str())
        } else {
            None
        }
    }
}
#[derive(Debug, Serialize, Deserialize)]
struct Remote_ {
    ssh: Option<SshRemote>,
    local: Option<String>,
    url: Option<String>,
}
#[derive(Debug)]
pub enum Remote {
    Ssh(SshRemote),
    Local { local: String },
    Http { url: String },
    None,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SshRemote {
    pub addr: String,
}
impl<'de> serde::Deserialize<'de> for Remote {
    fn deserialize<D>(deserializer: D) -> Result<Remote, D::Error>
    where
        D: serde::de::Deserializer<'de>,
    {
        let r = Remote_::deserialize(deserializer)?;
        if let Some(ssh) = r.ssh {
            Ok(Remote::Ssh(ssh))
        } else if let Some(local) = r.local {
            Ok(Remote::Local { local })
        } else if let Some(url) = r.url {
            Ok(Remote::Http { url })
        } else {
            Ok(Remote::None)
        }
    }
}
impl serde::Serialize for Remote {
    fn serialize<D>(&self, serializer: D) -> Result<D::Ok, D::Error>
    where
        D: serde::ser::Serializer,
    {
        let r = match *self {
            Remote::Ssh(ref ssh) => Remote_ {
                ssh: Some(ssh.clone()),
                local: None,
                url: None,
            },
            Remote::Local { ref local } => Remote_ {
                local: Some(local.to_string()),
                ssh: None,
                url: None,
            },
            Remote::Http { ref url } => Remote_ {
                local: None,
                ssh: None,
                url: Some(url.to_string()),
            },
            Remote::None => Remote_ {
                local: None,
                ssh: None,
                url: None,
            },
        };
        r.serialize(serializer)
    }
}

use std::collections::HashMap;
use std::path::PathBuf;
use super::{make_changelist, parse_changelist};
use anyhow::{anyhow, bail};
use clap::Clap;
use libpijul::changestore::ChangeStore;
use libpijul::pristine::sanakirja::Txn;
use libpijul::*;
use log::debug;
use crate::repository::Repository;
#[derive(Clap, Debug)]
pub struct Unrecord {
    /// Set the repository where this command should run. Defaults to the first ancestor of the current directory that contains a `.pijul` directory.
    #[clap(long = "repository")]
    repo_path: Option<PathBuf>,
    /// Unrecord changes from this channel instead of the current channel
    #[clap(long = "channel")]
    channel: Option<String>,
    /// Also undo the changes in the working copy (preserving unrecorded changes if there are any)
    #[clap(long = "reset")]
    reset: bool,
    /// Show N changes in a text editor if no <change-id>s were given.
    /// Defaults to the value
    /// of `unrecord_changes` in your global configuration.
    #[clap(long = "show-changes", value_name = "N", conflicts_with("change-id"))]
    show_changes: Option<usize>,
    /// The hash of a change (unambiguous prefixes are accepted)
    #[clap(multiple = true)]
    change_id: Vec<String>,
}
impl Unrecord {
    pub fn run(self) -> Result<(), anyhow::Error> {
        let mut repo = Repository::find_root(self.repo_path)?;
        debug!("{:?}", repo.config);
        let channel_name = repo.config.get_current_channel(self.channel.as_ref());
        let mut txn = repo.pristine.mut_txn_begin();
        if let Some(mut channel) = txn.load_channel(channel_name)? {
            let pending_hash = if self.reset {
                super::pending(&mut txn, &mut channel, &mut repo)?
            } else {
                None
            };
            let mut hashes = Vec::new();
            if self.change_id.is_empty() {
                // No change ids were given, present a list for choosing
                // The number can be set in the global config or passed as a command-line option
                let number_of_changes = if let Some(n) = self.show_changes {
                    n
                } else {
                    let cfg = crate::config::Global::load()?;
                    cfg.unrecord_changes.ok_or_else(|| {
                        anyhow!(
                            "Can't determine how many changes to show. \
                            Please set the `unrecord_changes` option in \
                            your global config or run `pijul unrecord` \
                            with the `--show-changes` option."
                        )
                    })?
                };
                let hashes_ = txn
                    .reverse_log(&channel.borrow(), None)?
                    .map(|h| (h.unwrap().1).0)
                    .take(number_of_changes)
                    .collect::<Vec<_>>();
                let o = make_changelist(&repo.changes, &hashes_, "unrecord")?;
                for h in parse_changelist(&edit::edit_bytes(&o[..])?).iter() {
                    hashes.push((*h, txn.get_internal(*h)?.unwrap()))
                }
            } else {
                for c in self.change_id.iter() {
                    let (hash, cid) = txn.hash_from_prefix(c)?;
                    hashes.push((hash, cid))
                }
            };
            let channel_ = channel.borrow();
            let mut changes = Vec::new();
            for (hash, change_id) in hashes {
                let n = txn
                    .get_changeset(Txn::changes(&channel_), change_id)
                    .unwrap();
                changes.push((hash, change_id, n));
            }
            std::mem::drop(channel_);
            changes.sort_by(|a, b| b.2.cmp(&a.2));
            for (hash, change_id, _) in changes {
                let channel_ = channel.borrow();
                for p in txn.iter_revdep(change_id)? {
                    let (p, d) = p?;
                    if p < change_id {
                        continue;
                    } else if p > change_id {
                        break;
                    }
                    if txn.get_changeset(Txn::changes(&channel_), d)?.is_some() {
                        let dep = txn.get_external(d)?.unwrap();
                        if Some(dep) == pending_hash {
                            bail!(
                                "Cannot unrecord change {} because unrecorded changes depend on it",
                                hash.to_base32()
                            );
                        } else {
                            bail!(
                                "Cannot unrecord change {} because {} depend on it",
                                hash.to_base32(),
                                dep.to_base32()
                            );
                        }
                    }
                }
                std::mem::drop(channel_);
                txn.unrecord(&repo.changes, &mut channel, &hash)?;
            }
            if self.reset {
                txn.output_repository_no_pending(
                    &mut repo.working_copy,
                    &repo.changes,
                    &mut channel,
                    &mut HashMap::new(),
                    "",
                    true,
                    None,
                )?;
            }
            if let Some(h) = pending_hash {
                txn.unrecord(&repo.changes, &mut channel, &h)?;
                if cfg!(feature = "keep-changes") {
                    repo.changes.del_change(&h)?;
                }
            }
        }
        txn.commit()?;
        Ok(())
    }
}

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use anyhow::bail;
use canonical_path::CanonicalPathBuf;
use clap::Clap;
use libpijul::pristine::{ChangeId, ChannelMutTxnT, Position};
use libpijul::{ChannelTxnT, DepsTxnT, MutTxnT, MutTxnTExt, TxnT, TxnTExt};
use log::debug;
use crate::repository::Repository;
#[derive(Clap, Debug)]
pub struct Reset {
    /// Set the repository where this command should run. Defaults to the first ancestor of the current directory that contains a `.pijul` directory.
    #[clap(long = "repository")]
    pub repo_path: Option<PathBuf>,
    /// Reset the working copy to this channel, and change the current channel to this channel.
    #[clap(long = "channel")]
    pub channel: Option<String>,
    /// Print this file to the standard output, without modifying the repository (works for a single file only).
    #[clap(long = "dry-run")]
    pub dry_run: bool,
    /// Only reset these files
    pub files: Vec<PathBuf>,
}
impl Reset {
    pub fn run(self) -> Result<(), anyhow::Error> {
        self.reset(true)
    }
    pub fn switch(self) -> Result<(), anyhow::Error> {
        self.reset(false)
    }
    fn reset(self, overwrite_changes: bool) -> Result<(), anyhow::Error> {
        let has_repo_path = self.repo_path.is_some();
        let mut repo = Repository::find_root(self.repo_path)?;
        let mut txn = repo.pristine.mut_txn_begin();
        let channel_name = repo.config.get_current_channel(self.channel.as_ref());
        let repo_path = CanonicalPathBuf::canonicalize(&repo.path)?;
        let mut channel = if let Some(channel) = txn.load_channel(&channel_name)? {
            channel
        } else {
            bail!("No such channel: {:?}", channel_name)
        };
        if self.dry_run {
            if self.files.len() != 1 {
                bail!("reset --dry-run needs exactly one file");
            }
            let (pos, _ambiguous) = if has_repo_path {
                let root = std::fs::canonicalize(repo.path.join(&self.files[0]))?;
                let path = root.strip_prefix(&repo_path)?.to_str().unwrap();
                txn.follow_oldest_path(&repo.changes, &channel, &path)?
            } else {
                let mut root = crate::current_dir()?;
                root.push(&self.files[0]);
                let root = std::fs::canonicalize(&root)?;
                let path = root.strip_prefix(&repo_path)?.to_str().unwrap();
                txn.follow_oldest_path(&repo.changes, &channel, &path)?
            };
            txn.output_file(
                &repo.changes,
                &channel,
                pos,
                &mut libpijul::vertex_buffer::Writer::new(std::io::stdout()),
            )?;
        } else {
            let current_channel = repo.config.get_current_channel(None);
            if self.channel.as_deref() == Some(current_channel) {
                if !overwrite_changes {
                    return Ok(());
                }
            } else if self.channel.is_some() {
                if let Some(mut channel) = txn.load_channel(current_channel)? {
                    let mut state = libpijul::RecordBuilder::new();
                    txn.record(
                        &mut state,
                        libpijul::Algorithm::default(),
                        &mut channel,
                        &mut repo.working_copy,
                        &repo.changes,
                        "",
                    )?;
                    let rec = state.finish();
                    debug!("actions = {:?}", rec.actions);
                    if !rec.actions.is_empty() {
                        bail!("Cannot change channel, as there are unrecorded changes.")
                    }
                }
            }
            let now = std::time::Instant::now();
            if self.files.is_empty() {
                if self.channel.is_none() || self.channel.as_deref() == Some(current_channel) {
                    let last_modified = last_modified(&txn, &channel.borrow());
                    txn.output_repository_no_pending(
                        &mut repo.working_copy,
                        &repo.changes,
                        &mut channel,
                        &mut HashMap::new(),
                        "",
                        true,
                        Some(last_modified),
                    )?;
                    txn.touch_channel(&mut channel.borrow_mut(), None);
                    txn.commit()?;
                    return Ok(());
                }
                let mut inodes = HashSet::new();
                if let Some(cur) = txn.load_channel(current_channel)? {
                    let mut changediff = HashSet::new();
                    let (a, b, s) = libpijul::pristine::last_common_state(
                        &txn,
                        &cur.borrow(),
                        &channel.borrow(),
                    )?;
                    debug!("last common state {:?}", s);
                    changes_after(&txn, &cur.borrow(), a, &mut changediff, &mut inodes)?;
                    changes_after(&txn, &channel.borrow(), b, &mut changediff, &mut inodes)?;
                }
                if self.channel.is_some() {
                    repo.config.current_channel = self.channel;
                    repo.save_config()?;
                }
                for pos in inodes.iter() {
                    let (path, _) = libpijul::fs::find_path(
                        &repo.changes,
                        &txn,
                        &channel.borrow(),
                        false,
                        *pos,
                    )?;
                    debug!("resetting {:?}", path);
                    txn.output_repository_no_pending(
                        &mut repo.working_copy,
                        &repo.changes,
                        &mut channel,
                        &mut HashMap::new(),
                        &path,
                        true,
                        None,
                    )?;
                }
            } else {
                let mut done = HashMap::new();
                for root in self.files.iter() {
                    let root = std::fs::canonicalize(&root)?;
                    let path = root.strip_prefix(&repo_path)?.to_str().unwrap();
                    txn.output_repository_no_pending(
                        &mut repo.working_copy,
                        &repo.changes,
                        &mut channel,
                        &mut done,
                        &path,
                        true,
                        None,
                    )?;
                }
            }
            txn.commit()?;
            debug!("now = {:?}", now.elapsed());
        }
        let locks = libpijul::TIMERS.lock().unwrap();
        debug!(
            "retrieve: {:?}, graph: {:?}, output: {:?}",
            locks.alive_retrieve, locks.alive_graph, locks.alive_output,
        );
        Ok(())
    }
}
fn changes_after<T: ChannelTxnT + DepsTxnT>(
    txn: &T,
    chan: &T::Channel,
    from: u64,
    changediff: &mut HashSet<ChangeId>,
    inodes: &mut HashSet<Position<ChangeId>>,
) -> Result<(), anyhow::Error> {
    for x in libpijul::pristine::changeid_log(txn, chan, from)? {
        let (n, (u, _)) = x?;
        debug!("{:?} {:?} {:?}", n, u, from);
        if n <= from {
            continue;
        }
        if changediff.insert(u) {
            for y in txn.iter_rev_touched_files(u, None)? {
                let (uu, pos) = y?;
                if uu < u {
                    continue;
                } else if uu > u {
                    break;
                }
                inodes.insert(pos);
            }
        }
    }
    Ok(())
}
fn last_modified<T: ChannelTxnT>(_: &T, channel: &T::Channel) -> std::time::SystemTime {
    std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(T::last_modified(channel))
}

use std::collections::HashMap;
use std::io::Write;
use std::path::PathBuf;
use anyhow::bail;
use canonical_path::CanonicalPathBuf;
use chrono::Utc;
use clap::Clap;
use libpijul::change::*;
use libpijul::changestore::*;
use libpijul::pristine::ChannelMutTxnT;
use libpijul::{Base32, ChannelRef, MutTxnT, MutTxnTExt, TxnT, TxnTExt};
use log::{debug, error};
use serde_derive::{Deserialize, Serialize};
use thrussh_keys::PublicKeyBase64;
use crate::repository::*;
#[derive(Clap, Debug)]
pub struct Record {
    /// Record all paths that have changed
    #[clap(short = 'a', long = "all")]
    pub all: bool,
    /// Set the change message
    #[clap(short = 'm', long = "message")]
    pub message: Option<String>,
    /// Set the author field
    #[clap(long = "author")]
    pub author: Option<String>,
    /// Record the change in this channel instead of the current channel
    #[clap(long = "channel")]
    pub channel: Option<String>,
    /// Set the repository where this command should run. Defaults to the first ancestor of the current directory that contains a `.pijul` directory.
    #[clap(long = "repository")]
    pub repo_path: Option<PathBuf>,
    /// Set the timestamp field
    #[clap(long = "timestamp")]
    pub timestamp: Option<i64>,
    /// Sign the change with the default key
    #[clap(short = 'S')]
    pub sign: bool,
    #[clap(long = "tag")]
    pub tag: bool,
    /// Amend this change instead of creating a new change
    #[clap(long = "amend")]
    #[allow(clippy::option_option)]
    pub amend: Option<Option<String>>,
    /// Paths in which to record the changes
    pub prefixes: Vec<PathBuf>,
}
impl Record {
    pub async fn run(self) -> Result<(), anyhow::Error> {
        let mut repo = Repository::find_root(self.repo_path.clone())?;
        let mut stdout = std::io::stdout();
        let mut stderr = std::io::stderr();
        for h in repo.config.hooks.record.iter() {
            let mut proc = std::process::Command::new("bash")
                .current_dir(&repo.path)
                .args(&["-c", &h])
                .spawn()?;
            let status = proc.wait()?;
            if !status.success() {
                writeln!(stderr, "Hook {:?} exited with code {:?}", h, status)?;
                std::process::exit(status.code().unwrap_or(1))
            }
        }
        let mut txn = repo.pristine.mut_txn_begin();
        let mut channel =
            txn.open_or_create_channel(repo.config.get_current_channel(self.channel.as_ref()))?;
        let sign = self.sign;
        let header = if let Some(ref amend) = self.amend {
            let h = if let Some(ref hash) = amend {
                txn.hash_from_prefix(hash)?.0
            } else if let Some(h) = txn.reverse_log(&channel.borrow(), None)?.next() {
                (h?.1).0
            } else {
                return Ok(());
            };
            let header = if let Some(message) = self.message.clone() {
                ChangeHeader {
                    message,
                    ..repo.changes.get_header(&h)?
                }
            } else {
                repo.changes.get_header(&h)?
            };
            txn.unrecord(&repo.changes, &mut channel, &h)?;
            header
        } else {
            self.header()
        };
        let no_prefixes = self.prefixes.is_empty();
        let result = self.record(
            &mut txn,
            &mut channel,
            &mut repo.working_copy,
            &repo.changes,
            CanonicalPathBuf::canonicalize(&repo.path)?,
            header,
        )?;
        if let Some((mut change, updates, hash, oldest)) = result {
            let hash = hash.unwrap();
            if sign {
                let mut key_path = dirs_next::home_dir().unwrap().join(".ssh");
                if let Some((pk, signature)) = sign_hash(&mut key_path, hash).await? {
                    let sig = toml::Value::try_from(vec![Signature {
                        public_key: pk,
                        timestamp: change.header.timestamp,
                        signature,
                    }])?;
                    let mut toml = toml::map::Map::new();
                    toml.insert("signatures".to_string(), sig);
                    change.unhashed = Some(toml.into());
                    let hash2 = repo.changes.save_change(&change).unwrap();
                    assert_eq!(hash2, hash);
                }
            }
            txn.apply_local_change(&mut channel, &change, hash, &updates)?;
            writeln!(stdout, "Hash: {}", hash.to_base32())?;
            let oldest = if let Ok(t) = oldest.duration_since(std::time::SystemTime::UNIX_EPOCH) {
                t.as_secs() as u64
            } else {
                0
            };
            if no_prefixes {
                txn.touch_channel(&mut channel.borrow_mut(), Some(oldest));
            }
            txn.commit()?;
        } else {
            if no_prefixes {
                txn.touch_channel(&mut channel.borrow_mut(), None);
                txn.commit()?;
            }
            writeln!(stderr, "Nothing to record")?;
        }
        Ok(())
    }
    fn header(&self) -> ChangeHeader {
        let authors = if let Some(ref a) = self.author {
            vec![libpijul::change::Author {
                name: a.clone(),
                full_name: None,
                email: None,
            }]
        } else if let Ok(global) = crate::config::Global::load() {
            vec![global.author]
        } else {
            Vec::new()
        };
        ChangeHeader {
            message: self.message.clone().unwrap_or_else(String::new),
            authors,
            description: None,
            timestamp: if let Some(t) = self.timestamp {
                chrono::DateTime::from_utc(chrono::NaiveDateTime::from_timestamp(t, 0), chrono::Utc)
            } else {
                Utc::now()
            },
        }
    }
    fn fill_relative_prefixes(&mut self) -> Result<(), anyhow::Error> {
        let cwd = std::env::current_dir()?;
        for p in self.prefixes.iter_mut() {
            if p.is_relative() {
                *p = cwd.join(&p);
            }
        }
        Ok(())
    }
    fn record<T: TxnTExt + MutTxnTExt, C: ChangeStore>(
        mut self,
        txn: &mut T,
        channel: &mut ChannelRef<T>,
        working_copy: &mut libpijul::working_copy::FileSystem,
        changes: &C,
        repo_path: CanonicalPathBuf,
        header: ChangeHeader,
    ) -> Result<
        Option<(
            Change,
            HashMap<usize, libpijul::InodeUpdate>,
            Option<libpijul::Hash>,
            std::time::SystemTime,
        )>,
        anyhow::Error,
    > {
        let mut state = libpijul::RecordBuilder::new();
        if self.prefixes.is_empty() {
            txn.record(
                &mut state,
                libpijul::Algorithm::default(),
                channel,
                working_copy,
                changes,
                "",
            )?
        } else {
            self.fill_relative_prefixes()?;
            working_copy.record_prefixes(
                txn,
                channel,
                changes,
                &mut state,
                repo_path,
                &self.prefixes,
                num_cpus::get(),
            )?;
        }
        let mut rec = state.finish();
        if rec.actions.is_empty() {
            return Ok(None);
        }
        let actions = rec
            .actions
            .into_iter()
            .map(|rec| rec.globalize(txn).unwrap())
            .collect();
        let change =
            LocalChange::make_change(txn, channel, actions, rec.contents, header, Vec::new())?;
        let file_name = |local: &Local, _| -> String { format!("{}:{}", local.path, local.line) };
        debug!("has_binary = {:?}", rec.has_binary_files);
        let change = if self.all {
            change
        } else if rec.has_binary_files {
            bail!("Cannot record a binary change interactively. Please use -a.")
        } else {
            let mut o = Vec::new();
            change.write(changes, None, file_name, true, &mut o)?;
            let mut with_errors: Option<Vec<u8>> = None;
            let change = loop {
                let mut bytes = if let Some(ref o) = with_errors {
                    edit::edit_bytes(&o[..])?
                } else {
                    edit::edit_bytes(&o[..])?
                };
                if bytes.iter().all(|c| (*c as char).is_whitespace()) {
                    return Ok(None);
                }
                let mut change = std::io::BufReader::new(std::io::Cursor::new(&bytes));
                if let Ok(change) =
                    Change::read_and_deps(&mut change, &mut rec.updatables, txn, channel)
                {
                    break change;
                }
                let mut err = SYNTAX_ERROR.as_bytes().to_vec();
                err.append(&mut bytes);
                with_errors = Some(err)
            };
            if change.changes.is_empty() {
                return Ok(None);
            }
            change
        };
        if change.header.message.trim().is_empty() {
            bail!("No change message")
        }
        debug!("saving change");
        let hash = changes.save_change(&change).unwrap();
        debug!("saved");
        Ok(Some((
            change,
            rec.updatables,
            Some(hash),
            rec.oldest_change,
        )))
    }
}
#[derive(Debug, Serialize, Deserialize)]
struct Signature {
    public_key: String,
    timestamp: chrono::DateTime<chrono::Utc>,
    signature: String,
}
async fn sign_hash(
    key_path: &mut PathBuf,
    hash: libpijul::Hash,
) -> Result<Option<(String, String)>, anyhow::Error> {
    let to_sign = hash.to_bytes();
    match thrussh_keys::agent::client::AgentClient::connect_env().await {
        Ok(agent) => {
            let mut agent = Some(agent);
            for k in &["id_ed25519.pub", "id_rsa.pub"] {
                key_path.push(k);
                if let Ok(key) = thrussh_keys::load_public_key(&key_path) {
                    debug!("key");
                    if let Some(a) = agent.take() {
                        debug!("authenticate future");
                        if let (_, Ok(sig)) = a.sign_request_base64(&key, &to_sign).await {
                            key_path.pop();
                            let key = key.public_key_base64();
                            return Ok(Some((key, sig)));
                        }
                    }
                }
                key_path.pop();
            }
        }
        Err(e) => {
            error!("{:?}", e);
        }
    }
    for k in &["id_ed25519", "id_rsa"] {
        key_path.push(k);
        if let Some(k) = crate::remote::ssh::load_secret_key(&key_path, k) {
            key_path.pop();
            let pk = k.public_key_base64();
            return Ok(Some((pk, k.sign_detached(&to_sign)?.to_base64())));
        } else {
            key_path.pop();
        }
    }
    Ok(None)
}
const SYNTAX_ERROR: &str = "# Syntax errors, please try again.
# Alternatively, you may delete the entire file (including this
# comment to abort).
";

use std::collections::{HashMap, HashSet};
use std::io::Write;
use std::path::PathBuf;
use super::{make_changelist, parse_changelist};
use anyhow::bail;
use clap::Clap;
use lazy_static::lazy_static;
use libpijul::changestore::ChangeStore;
use libpijul::pristine::sanakirja::Txn;
use libpijul::*;
use log::debug;
use regex::Regex;
use crate::repository::Repository;
#[derive(Clap, Debug)]
pub struct Remote {
    #[clap(subcommand)]
    subcmd: Option<SubRemote>,
    /// Set the repository where this command should run. Defaults to the first ancestor of the current directory that contains a `.pijul` directory.
    #[clap(long = "repository")]
    repo_path: Option<PathBuf>,
}
#[derive(Clap, Debug)]
pub enum SubRemote {
    /// Deletes the remote
    #[clap(name = "delete")]
    Delete { remote: String },
}
impl Remote {
    pub fn run(self) -> Result<(), anyhow::Error> {
        let repo = Repository::find_root(self.repo_path)?;
        debug!("{:?}", repo.config);
        let mut stdout = std::io::stdout();
        match self.subcmd {
            None => {
                let txn = repo.pristine.txn_begin()?;
                for r in txn.iter_remotes("")? {
                    let r = r?;
                    writeln!(stdout, "  {}", r.name())?;
                }
            }
            Some(SubRemote::Delete { remote }) => {
                let mut txn = repo.pristine.mut_txn_begin();
                if !txn.drop_named_remote(&remote)? {
                    writeln!(std::io::stderr(), "Remote not found: {:?}", remote)?
                } else {
                    txn.commit()?;
                }
            }
        }
        Ok(())
    }
}
#[derive(Clap, Debug)]
pub struct Push {
    /// Path to the repository. Uses the current repository if the argument is omitted
    #[clap(long = "repository")]
    repo_path: Option<PathBuf>,
    /// Push from this channel instead of the default channel
    #[clap(long = "from-channel")]
    from_channel: Option<String>,
    /// Push all changes
    #[clap(long = "all", short = 'a', conflicts_with = "changes")]
    all: bool,
    /// Do not check certificates (HTTPS remotes only, this option might be dangerous)
    #[clap(short = 'k')]
    no_cert_check: bool,
    /// Push changes only relating to these paths
    #[clap(long = "path")]
    path: Vec<String>,
    /// Push to this remote
    to: Option<String>,
    /// Push to this remote channel instead of the remote's default channel
    #[clap(long = "to-channel")]
    to_channel: Option<String>,
    /// Push only these changes
    #[clap(last = true)]
    changes: Vec<String>,
}
#[derive(Clap, Debug)]
pub struct Pull {
    /// Set the repository where this command should run. Defaults to the first ancestor of the current directory that contains a `.pijul` directory.
    #[clap(long = "repository")]
    repo_path: Option<PathBuf>,
    /// Pull into this channel instead of the current channel
    #[clap(long = "to-channel")]
    to_channel: Option<String>,
    /// Push all changes
    #[clap(long = "all", short = 'a', conflicts_with = "changes")]
    all: bool,
    /// Do not check certificates (HTTPS remotes only, this option might be dangerous)
    #[clap(short = 'k')]
    no_cert_check: bool,
    /// Download full changes, even when not necessory
    #[clap(long = "full")]
    full: bool, // This can't be symmetric with push
    /// Only pull to these paths
    #[clap(long = "path")]
    path: Vec<String>,
    /// Pull from this remote
    from: Option<String>,
    /// Pull from this remote channel
    #[clap(long = "from-channel")]
    from_channel: Option<String>,
    /// Pull changes from the local repository, not necessarily from a channel
    #[clap(last = true)]
    changes: Vec<String>, // For local changes only, can't be symmetric.
}
lazy_static! {
    static ref CHANNEL: Regex = Regex::new(r#"([^:]*)(:(.*))?"#).unwrap();
}
impl Push {
    pub async fn run(self) -> Result<(), anyhow::Error> {
        let mut stderr = std::io::stderr();
        let repo = Repository::find_root(self.repo_path)?;
        debug!("{:?}", repo.config);
        let channel_name = repo.config.get_current_channel(self.from_channel.as_ref());
        let remote_name = if let Some(ref rem) = self.to {
            rem
        } else if let Some(ref def) = repo.config.default_remote {
            def
        } else {
            bail!("Missing remote");
        };
        let mut push_channel = None;
        let remote_channel = if let Some(ref c) = self.to_channel {
            let c = CHANNEL.captures(c).unwrap();
            push_channel = c.get(3).map(|x| x.as_str());
            let c = c.get(1).unwrap().as_str();
            if c.is_empty() {
                channel_name
            } else {
                c
            }
        } else {
            channel_name
        };
        debug!("remote_channel = {:?} {:?}", remote_channel, push_channel);
        let mut remote = repo
            .remote(
                Some(&repo.path),
                &remote_name,
                remote_channel,
                self.no_cert_check,
            )
            .await?;
        let mut txn = repo.pristine.mut_txn_begin();
        let remote_changes = remote.update_changelist(&mut txn, &self.path).await?;
        let channel = txn.open_or_create_channel(channel_name)?;
        let mut paths = HashSet::new();
        for path in self.path.iter() {
            let (p, ambiguous) = txn.follow_oldest_path(&repo.changes, &channel, path)?;
            if ambiguous {
                bail!("Ambiguous path: {:?}", path)
            }
            paths.insert(p);
            paths.extend(
                libpijul::fs::iter_graph_descendants(&txn, &channel.borrow().graph, p)?
                    .map(|x| x.unwrap()),
            );
        }
        let mut to_upload = Vec::new();
        for x in txn.reverse_log(&channel.borrow(), None)? {
            let (_, (h, m)) = x?;
            if let Some((_, ref remote_changes)) = remote_changes {
                if txn.remote_has_state(remote_changes, m)? {
                    break;
                }
                let h_int = txn.get_internal(h)?.unwrap();
                if !txn.remote_has_change(&remote_changes, h)? {
                    if paths.is_empty() {
                        to_upload.push(h)
                    } else {
                        for p in paths.iter() {
                            if txn.get_touched_files(*p, Some(h_int))?.is_some() {
                                to_upload.push(h);
                                break;
                            }
                        }
                    }
                }
            } else if let crate::remote::RemoteRepo::LocalChannel(ref remote_channel) = remote {
                if let Some(channel) = txn.load_channel(remote_channel)? {
                    let channel = channel.borrow();
                    let h_int = txn.get_internal(h)?.unwrap();
                    if txn.get_changeset(Txn::changes(&channel), h_int)?.is_none() {
                        if paths.is_empty() {
                            to_upload.push(h)
                        } else {
                            for p in paths.iter() {
                                if txn.get_touched_files(*p, Some(h_int))?.is_some() {
                                    to_upload.push(h);
                                    break;
                                }
                            }
                        }
                    }
                }
            }
        }
        debug!("to_upload = {:?}", to_upload);
        if to_upload.is_empty() {
            writeln!(stderr, "Nothing to push")?;
            return Ok(());
        }
        to_upload.reverse();
        let to_upload = if !self.changes.is_empty() {
            let mut u = Vec::new();
            let mut not_found = Vec::new();
            for change in self.changes.iter() {
                match txn.hash_from_prefix(change) {
                    Ok((hash, _)) => {
                        if to_upload.contains(&hash) {
                            u.push(hash);
                        }
                    }
                    Err(_) => {
                        if !not_found.contains(change) {
                            not_found.push(change.to_string());
                        }
                    }
                }
            }
            if !not_found.is_empty() {
                bail!("Changes not found: {:?}", not_found)
            }
            check_deps(&repo.changes, &to_upload, &u)?;
            u
        } else if self.all {
            to_upload
        } else {
            let mut o = make_changelist(&repo.changes, &to_upload, "push")?;
            let remote_changes = remote_changes.map(|x| x.1);
            loop {
                let d = parse_changelist(&edit::edit_bytes(&o[..])?);
                let comp = complete_deps(&txn, &remote_changes, &repo.changes, &to_upload, &d)?;
                if comp.len() == d.len() {
                    break comp;
                }
                o = make_changelist(&repo.changes, &comp, "push")?
            }
        };
        debug!("to_upload = {:?}", to_upload);
        if to_upload.is_empty() {
            writeln!(stderr, "Nothing to push")?;
            return Ok(());
        }
        remote
            .upload_changes(&mut txn, repo.changes_dir.clone(), push_channel, &to_upload)
            .await?;
        txn.commit()?;
        remote.finish().await?;
        Ok(())
    }
}
impl Pull {
    pub async fn run(self) -> Result<(), anyhow::Error> {
        let mut repo = Repository::find_root(self.repo_path)?;
        let mut txn = repo.pristine.mut_txn_begin();
        let channel_name = repo.config.get_current_channel(self.to_channel.as_ref());
        let mut channel = txn.open_or_create_channel(channel_name)?;
        debug!("{:?}", repo.config);
        let remote_name = if let Some(ref rem) = self.from {
            rem
        } else if let Some(ref def) = repo.config.default_remote {
            def
        } else {
            bail!("Missing remote")
        };
        let from_channel = if let Some(ref c) = self.from_channel {
            c
        } else {
            crate::DEFAULT_CHANNEL
        };
        let mut remote = repo
            .remote(
                Some(&repo.path),
                &remote_name,
                from_channel,
                self.no_cert_check,
            )
            .await?;
        debug!("downloading");
        let mut inodes: HashSet<libpijul::pristine::Position<libpijul::Hash>> = HashSet::new();
        let mut to_download = if self.changes.is_empty() {
            let remote_changes = remote.update_changelist(&mut txn, &self.path).await?;
            debug!("changelist done");
            let mut to_download = Vec::new();
            if let Some((inodes_, remote_changes)) = remote_changes {
                inodes.extend(inodes_.into_iter());
                for x in txn.iter_remote(&remote_changes.borrow().remote, 0)? {
                    let (h, m) = x?.1;
                    if txn.channel_has_state(&channel.borrow(), m)?.is_some() {
                        break;
                    } else if txn.get_revchanges(&channel, h)?.is_none() {
                        to_download.push(h)
                    }
                }
            } else if let crate::remote::RemoteRepo::LocalChannel(ref remote_channel) = remote {
                let mut inodes_ = HashSet::new();
                for path in self.path.iter() {
                    let (p, ambiguous) = txn.follow_oldest_path(&repo.changes, &channel, path)?;
                    if ambiguous {
                        bail!("Ambiguous path: {:?}", path)
                    }
                    inodes_.insert(p);
                    inodes_.extend(
                        libpijul::fs::iter_graph_descendants(&txn, &channel.borrow().graph, p)?
                            .map(|x| x.unwrap()),
                    );
                }
                inodes.extend(inodes_.iter().map(|x| libpijul::pristine::Position {
                    change: txn.get_external(x.change).unwrap().unwrap(),
                    pos: x.pos,
                }));
                if let Some(remote_channel) = txn.load_channel(remote_channel)? {
                    let remote_channel = remote_channel.borrow();
                    for x in txn.reverse_log(&remote_channel, None)? {
                        let (h, m) = x?.1;
                        if txn.channel_has_state(&channel.borrow(), m)?.is_some() {
                            break;
                        }
                        let h_int = txn.get_internal(h)?.unwrap();
                        if txn
                            .get_changeset(Txn::changes(&channel.borrow()), h_int)?
                            .is_none()
                        {
                            if inodes_.is_empty()
                                || inodes_.iter().any(|&inode| {
                                    txn.get_rev_touched_files(h_int, Some(inode))
                                        .unwrap()
                                        .is_some()
                                })
                            {
                                to_download.push(h)
                            }
                        }
                    }
                }
            }
            to_download
        } else {
            let r: Result<Vec<libpijul::Hash>, anyhow::Error> = self
                .changes
                .iter()
                .map(|h| Ok(txn.hash_from_prefix(h)?.0))
                .collect();
            r?
        };
        debug!("recording");
        let hash = super::pending(&mut txn, &mut channel, &mut repo)?;
        let to_download = remote
            .pull(
                &mut repo,
                &mut txn,
                &mut channel,
                &mut to_download,
                &inodes,
                self.all,
            )
            .await?;
        if to_download.is_empty() {
            let mut stderr = std::io::stderr();
            writeln!(stderr, "Nothing to pull")?;
            return Ok(());
        }
        if !self.all {
            let mut o = make_changelist(&repo.changes, &to_download, "pull")?;
            let d = loop {
                let d = parse_changelist(&edit::edit_bytes(&o[..])?);
                let comp = complete_deps(&txn, &None, &repo.changes, &to_download, &d)?;
                if comp.len() == d.len() {
                    break comp;
                }
                o = make_changelist(&repo.changes, &comp, "pull")?
            };
            let mut ws = libpijul::ApplyWorkspace::new();
            debug!("to_download = {:?}", to_download);
            let progress = indicatif::ProgressBar::new(d.len() as u64);
            progress.set_style(
                indicatif::ProgressStyle::default_spinner()
                    .template("  Applying changes    {wide_bar}  {pos}/{len}"),
            );
            for h in d.iter() {
                txn.apply_change_rec_ws(&repo.changes, &mut channel, *h, &mut ws)?;
                progress.inc(1);
            }
            progress.set_style(
                indicatif::ProgressStyle::default_bar()
                    .template("✓ Applying changes    {wide_bar} {pos}/{len}"),
            );
            progress.finish();
        }
        debug!("completing changes");
        remote
            .complete_changes(&repo, &txn, &mut channel, &to_download, self.full)
            .await?;
        remote.finish().await?;
        let progress = indicatif::ProgressBar::new_spinner();
        progress.set_style(
            indicatif::ProgressStyle::default_spinner().template("{spinner} Outputting repository"),
        );
        progress.enable_steady_tick(100);
        debug!("inodes = {:?}", inodes);
        let mut touched = HashSet::new();
        for d in to_download.iter() {
            if let Some(int) = txn.get_internal(*d)? {
                for inode in txn.iter_rev_touched(int)? {
                    let (int_, inode) = inode?;
                    if int_ < int {
                        continue;
                    } else if int_ > int {
                        break;
                    }
                    let ext = libpijul::pristine::Position {
                        change: txn.get_external(inode.change)?.unwrap(),
                        pos: inode.pos,
                    };
                    if inodes.is_empty() || inodes.contains(&ext) {
                        touched.insert(inode);
                    }
                }
            }
        }
        let mut done = HashMap::new();
        for i in touched {
            let (path, _) =
                libpijul::fs::find_path(&repo.changes, &txn, &channel.borrow(), false, i)?;
            debug!("path = {:?}", path);
            txn.output_repository_no_pending(
                &mut repo.working_copy,
                &repo.changes,
                &mut channel,
                &mut done,
                &path,
                true,
                None,
            )?;
        }
        progress.set_style(
            indicatif::ProgressStyle::default_spinner().template("✓ Outputting repository"),
        );
        progress.finish();
        if let Some(h) = hash {
            txn.unrecord(&repo.changes, &mut channel, &h)?;
            repo.changes.del_change(&h)?;
        }
        txn.commit()?;
        Ok(())
    }
}
fn complete_deps<T: TxnT, C: ChangeStore>(
    txn: &T,
    remote_changes: &Option<libpijul::RemoteRef<T>>,
    c: &C,
    original: &[libpijul::Hash],
    now: &[libpijul::Hash],
) -> Result<Vec<libpijul::Hash>, anyhow::Error> {
    let original_: HashSet<_> = original.iter().collect();
    let mut now_ = HashSet::with_capacity(original.len());
    let mut result = Vec::with_capacity(original.len());
    for &h in now {
        now_.insert(h);
        result.push(h);
    }
    let mut stack = now.to_vec();
    stack.reverse();
    while let Some(n) = stack.pop() {
        // check that all of `now`'s deps are in now or not in original
        for d in c.get_dependencies(&n)? {
            if let Some(ref rem) = remote_changes {
                if txn.remote_has_change(rem, d)? {
                    continue;
                }
            }
            if original_.get(&d).is_some() && now_.get(&d).is_none() {
                result.push(d);
                now_.insert(d);
                stack.push(d);
            }
        }
        if now_.insert(n) {
            result.push(n)
        }
    }
    Ok(result)
}
fn check_deps<C: ChangeStore>(
    c: &C,
    original: &[libpijul::Hash],
    now: &[libpijul::Hash],
) -> Result<(), anyhow::Error> {
    let original_: HashSet<_> = original.iter().collect();
    let now_: HashSet<_> = now.iter().collect();
    for n in now {
        // check that all of `now`'s deps are in now or not in original
        for d in c.get_dependencies(n)? {
            if original_.get(&d).is_some() && now_.get(&d).is_none() {
                bail!("Missing dependency: {:?}", n)
            }
        }
    }
    Ok(())
}