twfss/src/bin/client.rs

use std::fs::create_dir_all;
use std::sync::Arc;
use std::ffi::{OsStr, OsString};

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{UnixListener, UnixStream};
use tokio::stream::StreamExt;
use tokio::sync::{mpsc, Mutex};
use tokio::signal;

use twfss::{Database, FileTree, ClientSideSync, FileSystemWatcher, Error, SynchronizationError, paths};

mod cli;

fn config_dir() -> String {
    format!(
        "{}/.config/twfss",
        dirs::home_dir().unwrap().to_str().unwrap().to_owned()
    )
}

fn data_dir() -> String {
    format!(
        "{}/.local/share/twfss-client",
        dirs::home_dir().unwrap().to_str().unwrap().to_owned()
    )
}

fn db_path() -> String {
    format!("{}/client.db", data_dir())
}

struct SynchronizedDirectory {
    path: OsString,
    client_side_sync: ClientSideSync,
    file_system_watcher: FileSystemWatcher,
    last_error: Option<Error>,
}

impl SynchronizedDirectory {
    async fn open(database: Arc<Mutex<Database>>, directory: &OsStr, errors: mpsc::Sender<SynchronizationError>) -> Result<SynchronizedDirectory, Error> {
        let file_tree = Arc::new(Mutex::new(FileTree::open(database, directory)?));
        let client_side_sync = ClientSideSync::new(file_tree.clone(), errors.clone()).await;
        let file_system_watcher = FileSystemWatcher::new(file_tree, errors).await;
        Ok(SynchronizedDirectory{
            path: directory.to_owned(),
            client_side_sync,
            file_system_watcher,
            last_error: None,
        })
    }

    async fn unpause(&mut self) {
        self.client_side_sync.unpause().await;
        self.file_system_watcher.unpause().await;
    }

    async fn pause(&mut self) {
        self.client_side_sync.pause().await;
        self.file_system_watcher.pause().await;
    }
}

#[tokio::main]
async fn main() {
    // Create the data directories if necessary.
    create_dir_all(config_dir()).unwrap();
    create_dir_all(data_dir()).unwrap();

    // Check whether the client is already running, and exit if it is.
    // TODO

    let mut ctrl_c = Box::pin(signal::ctrl_c());
    let (errors_send, mut errors) = mpsc::channel(32);

    // We create the socket before starting synchronization, so that the unwrap() below does not
    // cause corruption.
    let mut listener = UnixListener::bind(paths::client_socket()).unwrap();
    let mut incoming = listener.incoming();

    // Initialize the existing synchronized directories.
    let mut directories = Vec::new();
    let db = Arc::new(Mutex::new(Database::create_or_open(&db_path()).unwrap()));
    for directory in db.lock().await.synchronized_directories() {
        // The function cannot fail - we just fetched the directory from the database.
        let mut sync_dir = SynchronizedDirectory::open(db.clone(), &directory, errors_send.clone()).await.unwrap();
        sync_dir.unpause().await;
        directories.push(sync_dir);
    }

    loop {
        tokio::select! {
            result = &mut ctrl_c => {
                result.expect("could not listen for Ctrl+C");
                println!("Ctrl+C pressed, terminating...");
                // Ctrl+C was pressed, so we terminate the program.
                break;
            }
            stream = incoming.next() => {
                match stream {
                    Some(Ok(stream)) => {
                        let mut stream = stream;
                        match handle_cli_client(&mut stream, &mut directories, db.clone()).await {
                            Ok(()) => {}
                            Err(e) => {
                                // Log error and try to send it to the stream.
                                // TODO: We want to send a return code as well.
                                stream
                                    .write_all(format!("Error: {:?}", e).as_bytes())
                                    .await
                                    .ok();
                            }
                        };
                    }
                    Some(Err(err)) => {
                        eprintln!("Error while listening on the local unix socket: {:?}", err);
                        break;
                    }
                    None => {
                        eprintln!("listening for CLI connections failed");
                        break;
                    }
                }
            }
            error = errors.next() => {
                let error = error.unwrap();
                eprintln!("error: {:?}", error.error);
                for directory in directories.iter_mut() {
                    if directory.path == error.directory {
                        // In theory, there could be a race condition here where we pause a
                        // directory that was just recreated, when the directory which caused the
                        // error was already deleted.
                        if error.needs_pause {
                            directory.pause().await;
                        }
                        directory.last_error = Some(error.error);
                        break;
                    }
                }
            }
        }
    }

    // For a clean shutdown, we need to pause all synchtonized directories. Then,
    // we can be sure that the database is not modified anymore.
    for mut directory in directories.into_iter() {
        directory.pause().await;
    }
}

async fn handle_cli_client(stream: &mut UnixStream, directories: &mut Vec<SynchronizedDirectory>, _db: Arc<Mutex<Database>>) -> Result<(), Error> {
    let mut request = String::new();
    stream.read_to_string(&mut request).await?;

    let options: cli::Options = serde_json::from_str(&request)?;
    let verbose = options.verbose;
    match options.command {
        _cmd @ cli::Command::ListDirectories { .. } => {
            // TODO
        }
        _cmd @ cli::Command::AddDirectory { .. } => {
            // TODO
        }
        _cmd @ cli::Command::RemoveDirectory { .. } => {
            // TODO
        }
        cli::Command::Pause => {
            for  directory in directories.iter_mut() {
                directory.pause().await;
                if verbose {
                    stream.write_all(format!("Paused {}.\n", directory.path.to_string_lossy()).as_bytes()).await.ok();
                }
            }
            stream.write_all("Synchronization paused.\n".as_bytes()).await.ok();
        }
        cli::Command::Resume => {
            for  directory in directories.iter_mut() {
                directory.unpause().await;
                if verbose {
                    stream.write_all(format!("Resumed {}.\n", directory.path.to_string_lossy()).as_bytes()).await.ok();
                }
            }
            stream.write_all("Synchronization resumed.\n".as_bytes()).await.ok();
        }
    }

    Ok(())
}