mtxspooler/src/spool.rs

use std::collections::*;
use std::path::PathBuf;
use std::sync::Arc;

use futures::compat::Stream01CompatExt;
use futures::prelude::*;

use hyper::client::connect::dns::GaiResolver;
use hyper::client::connect::HttpConnector;
use hyper_tls::HttpsConnector;

use inotify::ffi::*;

use tokio::fs as tokiofs;
use tokio::prelude::*;

use tokio_inotify;

use url::Url;

use ruma_client::Client;
use ruma_client_api::r0::message as rumamessage;
use ruma_events::{self, room::message::*};
use ruma_identifiers::RoomId;

use crate::config::*;

type MatrixClient = Client<HttpsConnector<HttpConnector<GaiResolver>>>;
type MessageRequest = rumamessage::create_message_event::Request;

#[derive(Debug)]
pub enum Error {
    FileFormatMismatch,
    BadUrl,
    Io(io::Error),
    MtxClient(ruma_client::Error),
}

impl From<io::Error> for Error {
    fn from(f: io::Error) -> Self {
        Self::Io(f)
    }
}

impl From<ruma_client::Error> for Error {
    fn from(f: ruma_client::Error) -> Self {
        Self::MtxClient(f)
    }
}

#[derive(Clone)]
struct SpoolAction {
    client: Arc<MatrixClient>,
    room: RoomId,
    delay_secs: u32,
    watch_path: PathBuf,
}

pub async fn start_spooling(conf: Config) -> Result<(), Error> {
    let ain = tokio_inotify::AsyncINotify::init().expect("inotify init");

    let mut clients = HashMap::new();
    let mut watch_map = HashMap::new();

    for s in conf.spool_dirs {
        let label = s.sender_acct_label;
        let w = ain.add_watch(&s.path, IN_CLOSE_WRITE | IN_MOVED_TO)?;
        println!("Added watch: {}", s.path.to_str().unwrap_or("[non-UTF-8]"));

        // TODO Make this better.
        let cli = if !clients.contains_key(&label) {
            let acc = conf.accounts.get(&label).expect("missing account");
            let cli = create_and_auth_client(acc.clone()).await?;
            clients.insert(label, cli.clone());
            cli
        } else {
            clients.get(&label).expect("missing account").clone()
        };

        let sa = SpoolAction {
            client: cli,
            room: s.dest_room_id,
            delay_secs: s.send_delay_sec,
            watch_path: s.path.clone(),
        };

        watch_map.insert(w, sa);
    }

    let watch_map = Arc::new(watch_map);

    // This is horrible.
    let _ = ain
        .compat()
        .map_err(Error::from)
        .try_for_each({
            |e| {
                // Not sure why we have to do this clone here outside.
                let wm = watch_map.clone();
                async move {
                    // I don't like these clones but idk any better way.
                    let act = match wm.as_ref().get(&e.wd) {
                        Some(a) => a.clone(),
                        None => {
                            println!(
                                "got a wd that was not from a watch we added, ignoring: {:?}",
                                e.wd
                            );
                            return Ok(());
                        }
                    };

                    //let pb = e.name.clone();

                    // TODO This should be spawning a new task.
                    //tokio::spawn(async move {
                    // TODO Respect delay.
                    match process_file(&e.name, &act).await {
                        Ok(()) => {} // ok I guess?
                        Err(e) => println!("Error processing file: {:?}", e),
                    }
                    //});

                    Ok(())
                }
            }
        })
        .await?;

    Ok(())
}

async fn process_file(p: &PathBuf, sa: &SpoolAction) -> Result<(), Error> {
    let ext = match p.extension().map(|e| e.to_str()).flatten() {
        Some(v) => v,
        None => {
            println!("Found weird file {:?}, ignoring", p);
            return Ok(());
        }
    };

    let name = p
        .file_name()
        .map(|e| e.to_str())
        .flatten()
        .unwrap_or("[non-UTF-8]");

    // This makes me *mad*.
    let mut real_path = sa.watch_path.clone();
    real_path.push(p);

    match ext {
        "txt" => {
            println!("Processing file for {} at {:?}", sa.room, p);

            let buf = match file_as_string(&real_path).await {
                Ok(v) => v,
                Err(Error::FileFormatMismatch) => {
                    println!("File {} is not UTF-8, ignoring", name);
                    return Ok(());
                }
                Err(e) => return Err(e),
            };

            let mut rng = rand::thread_rng();
            let req = make_text_request(sa.room.clone(), buf.as_str(), &mut rng);
            match sa.client.as_ref().request(req).await {
                Ok(_) => {
                    // Now delete it if it passed.
                    tokiofs::remove_file(real_path).await?;
                }
                Err(e) => println!("Error processing {}: {:?}", name, e),
            }
        }
        _ => println!(
            "Found file {:?}, but it has unsupported extension \"{}\"",
            p, ext
        ),
    }

    Ok(())
}

async fn file_as_string(p: &PathBuf) -> Result<String, Error> {
    let mut buf = Vec::new();
    let mut f = tokiofs::File::open(p).await?;
    f.read_to_end(&mut buf).await?;
    String::from_utf8(buf).map_err(|_| Error::FileFormatMismatch)
}

async fn create_and_auth_client(acct: Account) -> Result<Arc<MatrixClient>, Error> {
    let hs_url = Url::parse(&acct.homeserver).map_err(|_| Error::BadUrl)?;
    let c = MatrixClient::https(hs_url, None);
    match acct.auth {
        Auth::UsernamePass(un, pw) => c.log_in(un, pw, acct.device_id, acct.display).await?,
    };
    Ok(Arc::new(c))
}

fn make_text_request<R: rand::Rng>(room_id: RoomId, msg: &str, rng: &mut R) -> MessageRequest {
    let inner = TextMessageEventContent {
        body: String::from(msg),
        format: None,
        formatted_body: None,
        relates_to: None,
    };
    let mec = MessageEventContent::Text(inner);
    MessageRequest {
        room_id: room_id,
        event_type: ruma_events::EventType::RoomMessage,
        txn_id: make_txn_id(rng),
        data: mec,
    }
}

const TXN_ID_LEN: usize = 20;

fn make_txn_id<R: rand::Rng>(rng: &mut R) -> String {
    let mut buf = String::with_capacity(TXN_ID_LEN);
    for _ in 0..TXN_ID_LEN {
        buf.push((rng.gen_range(0, 26) + ('a' as u8)) as char);
    }
    buf
}