troll-patrol/src/bin/server.rs

use troll_patrol::{request_handler::handle, *};

use clap::Parser;
use futures::future;
use hyper::{
    server::conn::AddrStream,
    service::{make_service_fn, service_fn},
    Body, Request, Response, Server,
};
use serde::Deserialize;
use sled::Db;
use std::{
    collections::BTreeMap, convert::Infallible, fs::File, io::BufReader, net::SocketAddr,
    path::PathBuf, time::Duration,
};
use tokio::{
    signal, spawn,
    sync::{broadcast, mpsc, oneshot},
    time::sleep,
};
use tokio_cron::{Job, Scheduler};

async fn shutdown_signal() {
    tokio::signal::ctrl_c()
        .await
        .expect("failed to listen for ctrl+c signal");
    println!("Shut down Troll Patrol Server");
}

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// Name/path of the configuration file
    #[arg(short, long, default_value = "config.json")]
    config: PathBuf,
}

#[derive(Debug, Deserialize)]
pub struct Config {
    pub db: DbConfig,
    // map of distributor name to IP:port to contact it
    pub distributors: BTreeMap<BridgeDistributor, String>,
    extra_infos_base_url: String,
    // confidence required to consider a bridge blocked
    confidence: f64,
    //require_bridge_token: bool,
    port: u16,
    updater_schedule: String,
}

#[derive(Debug, Deserialize)]
pub struct DbConfig {
    // The path for the server database, default is "server_db"
    pub db_path: String,
}

impl Default for DbConfig {
    fn default() -> DbConfig {
        DbConfig {
            db_path: "server_db".to_owned(),
        }
    }
}

async fn update_daily_info(
    db: &Db,
    distributors: &BTreeMap<BridgeDistributor, String>,
    extra_infos_base_url: &str,
    confidence: f64,
) {
    update_extra_infos(&db, &extra_infos_base_url)
        .await
        .unwrap();
    update_negative_reports(&db, &distributors).await;
    update_positive_reports(&db, &distributors).await;
    let new_blockages = guess_blockages(&db, &analyzer::ExampleAnalyzer {}, confidence);
    report_blockages(&distributors, new_blockages).await;
}

async fn run_updater(updater_tx: mpsc::Sender<Command>) {
    updater_tx.send(Command::Update {}).await.unwrap();
}

async fn create_context_manager(
    db_config: DbConfig,
    distributors: BTreeMap<BridgeDistributor, String>,
    extra_infos_base_url: &str,
    confidence: f64,
    context_rx: mpsc::Receiver<Command>,
    mut kill: broadcast::Receiver<()>,
) {
    tokio::select! {
        create_context = context_manager(db_config, distributors, extra_infos_base_url, confidence, context_rx) => create_context,
        _ = kill.recv() => {println!("Shut down manager");},
    }
}

async fn context_manager(
    db_config: DbConfig,
    distributors: BTreeMap<BridgeDistributor, String>,
    extra_infos_base_url: &str,
    confidence: f64,
    mut context_rx: mpsc::Receiver<Command>,
) {
    let db: Db = sled::open(&db_config.db_path).unwrap();

    while let Some(cmd) = context_rx.recv().await {
        use Command::*;
        match cmd {
            Request { req, sender } => {
                let response = handle(&db, req).await;
                if let Err(e) = sender.send(response) {
                    eprintln!("Server Response Error: {:?}", e);
                };
                sleep(Duration::from_millis(1)).await;
            }
            Shutdown { shutdown_sig } => {
                println!("Sending Shutdown Signal, all threads should shutdown.");
                drop(shutdown_sig);
                println!("Shutdown Sent.");
            }
            Update {} => {
                update_daily_info(&db, &distributors, &extra_infos_base_url, confidence).await;
            }
        }
    }
}

// Each of the commands that can be handled
#[derive(Debug)]
enum Command {
    Request {
        req: Request<Body>,
        sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
    },
    Shutdown {
        shutdown_sig: broadcast::Sender<()>,
    },
    Update {},
}

#[tokio::main]
async fn main() {
    let args: Args = Args::parse();

    let config: Config = serde_json::from_reader(BufReader::new(
        File::open(&args.config).expect("Could not read config file"),
    ))
    .expect("Reading config file from JSON failed");

    let (request_tx, request_rx) = mpsc::channel(32);

    let updater_tx = request_tx.clone();
    let shutdown_cmd_tx = request_tx.clone();

    // create the shutdown broadcast channel and clone for every thread
    let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16);
    let kill = shutdown_tx.subscribe();
    // TODO: Gracefully shut down updater
    let kill_updater = shutdown_tx.subscribe();

    // Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx
    let shutdown_handler = spawn(async move {
        tokio::select! {
            _ = signal::ctrl_c() => {
                let cmd = Command::Shutdown {
                    shutdown_sig: shutdown_tx,
                };
                shutdown_cmd_tx.send(cmd).await.unwrap();
                sleep(Duration::from_secs(1)).await;

                _ = shutdown_rx.recv().await;
            }
        }
    });

    let updater = spawn(async move {
        // Run updater once per day
        let mut sched = Scheduler::utc();
        sched.add(Job::new(config.updater_schedule, move || {
            run_updater(updater_tx.clone())
        }));
    });

    let context_manager = spawn(async move {
        create_context_manager(
            config.db,
            config.distributors,
            &config.extra_infos_base_url,
            config.confidence,
            request_rx,
            kill,
        )
        .await
    });

    let make_service = make_service_fn(move |_conn: &AddrStream| {
        let request_tx = request_tx.clone();
        let service = service_fn(move |req| {
            let request_tx = request_tx.clone();
            let (response_tx, response_rx) = oneshot::channel();
            let cmd = Command::Request {
                req,
                sender: response_tx,
            };
            async move {
                request_tx.send(cmd).await.unwrap();
                response_rx.await.unwrap()
            }
        });
        async move { Ok::<_, Infallible>(service) }
    });

    let addr = SocketAddr::from(([0, 0, 0, 0], config.port));
    let server = Server::bind(&addr).serve(make_service);
    let graceful = server.with_graceful_shutdown(shutdown_signal());
    println!("Listening on {}", addr);
    if let Err(e) = graceful.await {
        eprintln!("server error: {}", e);
    }
    future::join_all([context_manager, updater, shutdown_handler]).await;
}
Add functions to verify reports, update db 2024-03-23 17:43:18 -04:00			`use troll_patrol::{request_handler::handle, *};`
Main method to download and process extra-infos 2024-02-07 18:42:08 -05:00
			`use clap::Parser;`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`use futures::future;`
			`use hyper::{`
			`server::conn::AddrStream,`
			`service::{make_service_fn, service_fn},`
			`Body, Request, Response, Server,`
			`};`
			`use serde::Deserialize;`
Main method to download and process extra-infos 2024-02-07 18:42:08 -05:00			`use sled::Db;`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`use std::{`
Add functions to verify reports, update db 2024-03-23 17:43:18 -04:00			`collections::BTreeMap, convert::Infallible, fs::File, io::BufReader, net::SocketAddr,`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`path::PathBuf, time::Duration,`
			`};`
			`use tokio::{`
			`signal, spawn,`
			`sync::{broadcast, mpsc, oneshot},`
			`time::sleep,`
			`};`
Run updater daily 2024-03-25 20:41:34 -04:00			`use tokio_cron::{Job, Scheduler};`
Main method to download and process extra-infos 2024-02-07 18:42:08 -05:00
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`async fn shutdown_signal() {`
			`tokio::signal::ctrl_c()`
			`.await`
			`.expect("failed to listen for ctrl+c signal");`
			`println!("Shut down Troll Patrol Server");`
			`}`
Main method to download and process extra-infos 2024-02-07 18:42:08 -05:00
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`#[derive(Parser, Debug)]`
			`#[command(author, version, about, long_about = None)]`
			`struct Args {`
			`/// Name/path of the configuration file`
			`#[arg(short, long, default_value = "config.json")]`
			`config: PathBuf,`
			`}`
Main method to download and process extra-infos 2024-02-07 18:42:08 -05:00
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`#[derive(Debug, Deserialize)]`
			`pub struct Config {`
			`pub db: DbConfig,`
Add functions to verify reports, update db 2024-03-23 17:43:18 -04:00			`// map of distributor name to IP:port to contact it`
			`pub distributors: BTreeMap<BridgeDistributor, String>,`
Store bridge data by fingerprint and store set of fingerprints We had been storing all the data as a huge map under the key 'bridges' so that we could iterate over all keys in the map as a way of iterating over all bridges. This caused the database to grow way too big, so we don't do that anymore. Now, we store each bridge's data with the bridge fingerprint as the key, and we store a set of bridge fingerprints under the key 'bridges'. 2024-04-06 01:09:43 -04:00			`extra_infos_base_url: String,`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`// confidence required to consider a bridge blocked`
			`confidence: f64,`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`//require_bridge_token: bool,`
			`port: u16,`
Operator supplies updater schedule in config 2024-04-12 12:32:01 -04:00			`updater_schedule: String,`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`}`

			`#[derive(Debug, Deserialize)]`
			`pub struct DbConfig {`
			`// The path for the server database, default is "server_db"`
			`pub db_path: String,`
			`}`

			`impl Default for DbConfig {`
			`fn default() -> DbConfig {`
			`DbConfig {`
			`db_path: "server_db".to_owned(),`
			`}`
			`}`
			`}`

Store bridge data by fingerprint and store set of fingerprints We had been storing all the data as a huge map under the key 'bridges' so that we could iterate over all keys in the map as a way of iterating over all bridges. This caused the database to grow way too big, so we don't do that anymore. Now, we store each bridge's data with the bridge fingerprint as the key, and we store a set of bridge fingerprints under the key 'bridges'. 2024-04-06 01:09:43 -04:00			`async fn update_daily_info(`
			`db: &Db,`
			`distributors: &BTreeMap<BridgeDistributor, String>,`
			`extra_infos_base_url: &str,`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`confidence: f64,`
Store bridge data by fingerprint and store set of fingerprints We had been storing all the data as a huge map under the key 'bridges' so that we could iterate over all keys in the map as a way of iterating over all bridges. This caused the database to grow way too big, so we don't do that anymore. Now, we store each bridge's data with the bridge fingerprint as the key, and we store a set of bridge fingerprints under the key 'bridges'. 2024-04-06 01:09:43 -04:00			`) {`
Don't persist extra-infos files to disk, just process and store results 2024-04-06 12:02:53 -04:00			`update_extra_infos(&db, &extra_infos_base_url)`
			`.await`
			`.unwrap();`
Add functions to verify reports, update db 2024-03-23 17:43:18 -04:00			`update_negative_reports(&db, &distributors).await;`
			`update_positive_reports(&db, &distributors).await;`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`let new_blockages = guess_blockages(&db, &analyzer::ExampleAnalyzer {}, confidence);`
Store bridge data with key 'bridges', guess and report blockages 2024-03-23 23:16:20 -04:00			`report_blockages(&distributors, new_blockages).await;`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`}`

Run updater daily 2024-03-25 20:41:34 -04:00			`async fn run_updater(updater_tx: mpsc::Sender<Command>) {`
			`updater_tx.send(Command::Update {}).await.unwrap();`
			`}`

Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`async fn create_context_manager(`
			`db_config: DbConfig,`
Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`distributors: BTreeMap<BridgeDistributor, String>,`
Store bridge data by fingerprint and store set of fingerprints We had been storing all the data as a huge map under the key 'bridges' so that we could iterate over all keys in the map as a way of iterating over all bridges. This caused the database to grow way too big, so we don't do that anymore. Now, we store each bridge's data with the bridge fingerprint as the key, and we store a set of bridge fingerprints under the key 'bridges'. 2024-04-06 01:09:43 -04:00			`extra_infos_base_url: &str,`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`confidence: f64,`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`context_rx: mpsc::Receiver<Command>,`
			`mut kill: broadcast::Receiver<()>,`
			`) {`
			`tokio::select! {`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`create_context = context_manager(db_config, distributors, extra_infos_base_url, confidence, context_rx) => create_context,`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`_ = kill.recv() => {println!("Shut down manager");},`
			`}`
			`}`

Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`async fn context_manager(`
			`db_config: DbConfig,`
			`distributors: BTreeMap<BridgeDistributor, String>,`
Store bridge data by fingerprint and store set of fingerprints We had been storing all the data as a huge map under the key 'bridges' so that we could iterate over all keys in the map as a way of iterating over all bridges. This caused the database to grow way too big, so we don't do that anymore. Now, we store each bridge's data with the bridge fingerprint as the key, and we store a set of bridge fingerprints under the key 'bridges'. 2024-04-06 01:09:43 -04:00			`extra_infos_base_url: &str,`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`confidence: f64,`
Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`mut context_rx: mpsc::Receiver<Command>,`
			`) {`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`let db: Db = sled::open(&db_config.db_path).unwrap();`

			`while let Some(cmd) = context_rx.recv().await {`
			`use Command::*;`
			`match cmd {`
			`Request { req, sender } => {`
			`let response = handle(&db, req).await;`
			`if let Err(e) = sender.send(response) {`
			`eprintln!("Server Response Error: {:?}", e);`
			`};`
			`sleep(Duration::from_millis(1)).await;`
			`}`
			`Shutdown { shutdown_sig } => {`
			`println!("Sending Shutdown Signal, all threads should shutdown.");`
			`drop(shutdown_sig);`
			`println!("Shutdown Sent.");`
			`}`
Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`Update {} => {`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`update_daily_info(&db, &distributors, &extra_infos_base_url, confidence).await;`
Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`}`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`}`
			`}`
			`}`

			`// Each of the commands that can be handled`
			`#[derive(Debug)]`
			`enum Command {`
			`Request {`
			`req: Request<Body>,`
			`sender: oneshot::Sender<Result<Response<Body>, Infallible>>,`
			`},`
			`Shutdown {`
			`shutdown_sig: broadcast::Sender<()>,`
			`},`
Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`Update {},`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`}`

			`#[tokio::main]`
			`async fn main() {`
			`let args: Args = Args::parse();`

			`let config: Config = serde_json::from_reader(BufReader::new(`
			`File::open(&args.config).expect("Could not read config file"),`
			`))`
			`.expect("Reading config file from JSON failed");`

			`let (request_tx, request_rx) = mpsc::channel(32);`

Run updater daily 2024-03-25 20:41:34 -04:00			`let updater_tx = request_tx.clone();`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00			`let shutdown_cmd_tx = request_tx.clone();`

			`// create the shutdown broadcast channel and clone for every thread`
			`let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16);`
			`let kill = shutdown_tx.subscribe();`
Run updater daily 2024-03-25 20:41:34 -04:00			`// TODO: Gracefully shut down updater`
			`let kill_updater = shutdown_tx.subscribe();`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00
			`// Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx`
			`let shutdown_handler = spawn(async move {`
			`tokio::select! {`
			`_ = signal::ctrl_c() => {`
			`let cmd = Command::Shutdown {`
			`shutdown_sig: shutdown_tx,`
			`};`
			`shutdown_cmd_tx.send(cmd).await.unwrap();`
			`sleep(Duration::from_secs(1)).await;`

			`_ = shutdown_rx.recv().await;`
			`}`
			`}`
			`});`

Run updater daily 2024-03-25 20:41:34 -04:00			`let updater = spawn(async move {`
			`// Run updater once per day`
			`let mut sched = Scheduler::utc();`
Operator supplies updater schedule in config 2024-04-12 12:32:01 -04:00			`sched.add(Job::new(config.updater_schedule, move \|\| {`
Run updater daily 2024-03-25 20:41:34 -04:00			`run_updater(updater_tx.clone())`
			`}));`
			`});`

Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`let context_manager = spawn(async move {`
Store bridge data by fingerprint and store set of fingerprints We had been storing all the data as a huge map under the key 'bridges' so that we could iterate over all keys in the map as a way of iterating over all bridges. This caused the database to grow way too big, so we don't do that anymore. Now, we store each bridge's data with the bridge fingerprint as the key, and we store a set of bridge fingerprints under the key 'bridges'. 2024-04-06 01:09:43 -04:00			`create_context_manager(`
			`config.db,`
			`config.distributors,`
			`&config.extra_infos_base_url,`
Pass confidence to analyzer 2024-04-12 12:52:30 -04:00			`config.confidence,`
Store bridge data by fingerprint and store set of fingerprints We had been storing all the data as a huge map under the key 'bridges' so that we could iterate over all keys in the map as a way of iterating over all bridges. This caused the database to grow way too big, so we don't do that anymore. Now, we store each bridge's data with the bridge fingerprint as the key, and we store a set of bridge fingerprints under the key 'bridges'. 2024-04-06 01:09:43 -04:00			`request_rx,`
			`kill,`
			`)`
			`.await`
Add command to update data and report blockages 2024-03-25 19:37:00 -04:00			`});`
Listen for reports Downgrade hyper, listen for new reports, TODO: test 2024-03-22 23:42:22 -04:00
			`let make_service = make_service_fn(move \|_conn: &AddrStream\| {`
			`let request_tx = request_tx.clone();`
			`let service = service_fn(move \|req\| {`
			`let request_tx = request_tx.clone();`
			`let (response_tx, response_rx) = oneshot::channel();`
			`let cmd = Command::Request {`
			`req,`
			`sender: response_tx,`
			`};`
			`async move {`
			`request_tx.send(cmd).await.unwrap();`
			`response_rx.await.unwrap()`
			`}`
			`});`
			`async move { Ok::<_, Infallible>(service) }`
			`});`

			`let addr = SocketAddr::from(([0, 0, 0, 0], config.port));`
			`let server = Server::bind(&addr).serve(make_service);`
			`let graceful = server.with_graceful_shutdown(shutdown_signal());`
			`println!("Listening on {}", addr);`
			`if let Err(e) = graceful.await {`
			`eprintln!("server error: {}", e);`
			`}`
Run updater daily 2024-03-25 20:41:34 -04:00			`future::join_all([context_manager, updater, shutdown_handler]).await;`
Main method to download and process extra-infos 2024-02-07 18:42:08 -05:00			`}`