ptth/src/server/mod.rs

use std::{
	error::Error,
	path::PathBuf,
	sync::Arc,
	time::Duration,
};

use futures::FutureExt;
use handlebars::Handlebars;
use hyper::{
	StatusCode,
};
use reqwest::Client;
use serde::Deserialize;
use tokio::{
	sync::oneshot,
	time::delay_for,
};

use crate::{
	http_serde,
	prelude::*,
};

pub mod file_server;

struct ServerState {
	config: Config,
	handlebars: Handlebars <'static>,
	server_info: file_server::ServerInfo,
	client: Client,
	hidden_path: Option <PathBuf>,
}

async fn handle_req_resp <'a> (
	state: &Arc <ServerState>,
	req_resp: reqwest::Response
) -> Result <(), ()> {
	//println! ("Step 1");
	
	let body = req_resp.bytes ().await.unwrap ();
	let wrapped_reqs: Vec <http_serde::WrappedRequest> = match rmp_serde::from_read_ref (&body)
	{
		Ok (x) => x,
		Err (e) => {
			error! ("Can't parse wrapped requests: {:?}", e);
			return Err (());
		},
	};
	
	debug! ("Unwrapped {} requests", wrapped_reqs.len ());
	
	for wrapped_req in wrapped_reqs.into_iter () {
		let state = state.clone ();
		
		tokio::spawn (async move {
			let (req_id, parts) = (wrapped_req.id, wrapped_req.req);
			
			debug! ("Handling request {}", req_id);
			
			let default_root = PathBuf::from ("./");
			let file_server_root: &std::path::Path = state.config.file_server_root
			.as_ref ()
			.unwrap_or (&default_root);
			
			let response = file_server::serve_all (
				&state.handlebars, 
				&state.server_info,
				file_server_root,
				parts.method, 
				&parts.uri, 
				&parts.headers,
				state.hidden_path.as_ref ().map (|p| p.as_path ())
			).await;
			
			let mut resp_req = state.client
			.post (&format! ("{}/http_response/{}", state.config.relay_url, req_id))
			.header (crate::PTTH_MAGIC_HEADER, base64::encode (rmp_serde::to_vec (&response.parts).unwrap ()));
			
			if let Some (length) = response.content_length {
				resp_req = resp_req.header ("Content-Length", length.to_string ());
			}
			if let Some (body) = response.body {
				resp_req = resp_req.body (reqwest::Body::wrap_stream (body));
			}
			
			let req = resp_req.build ().unwrap ();
			
			debug! ("{:?}", req.headers ());
			
			//println! ("Step 6");
			match state.client.execute (req).await {
				Ok (r) => {
					let status = r.status ();
					let text = r.text ().await.unwrap ();
					debug! ("{:?} {:?}", status, text);
				},
				Err (e) => {
					if e.is_request () {
						warn! ("Error while POSTing response. Client probably hung up.");
					}
					else {
						error! ("Err: {:?}", e);
					}
				},
			}
			
		});
	}
	
	Ok (())
}

#[derive (Default, Deserialize)]
pub struct ConfigFile {
	pub name: String,
	pub api_key: String,
	pub relay_url: String,
	pub file_server_root: Option <PathBuf>,
}

#[derive (Default)]
pub struct Config {
	pub relay_url: String,
	pub file_server_root: Option <PathBuf>,
}

pub async fn run_server (
	config_file: ConfigFile,
	shutdown_oneshot: oneshot::Receiver <()>,
	hidden_path: Option <PathBuf>
)
-> Result <(), Box <dyn Error>> 
{
	use std::convert::TryInto;
	
	if crate::password_is_bad (config_file.api_key.clone ()) {
		panic! ("API key is too weak, server can't use it");
	}
	
	let server_info = file_server::ServerInfo {
		server_name: config_file.name.clone (),
	};
	
	let tripcode = base64::encode (blake3::hash (config_file.api_key.as_bytes ()).as_bytes ());
	
	info! ("Server name is {}", config_file.name);
	info! ("Tripcode is {}", tripcode);
	
	let mut headers = reqwest::header::HeaderMap::new ();
	headers.insert ("X-ApiKey", config_file.api_key.try_into ().unwrap ());
	
	let client = Client::builder ()
	.default_headers (headers)
	.timeout (Duration::from_secs (40))
	.build ().unwrap ();
	let handlebars = file_server::load_templates ()?;
	
	let state = Arc::new (ServerState {
		config: Config {
			relay_url: config_file.relay_url,
			file_server_root: config_file.file_server_root,
		},
		handlebars,
		server_info,
		client,
		hidden_path,
	});
	
	let mut backoff_delay = 0;
	let mut shutdown_oneshot = shutdown_oneshot.fuse ();
	
	loop {
		// TODO: Extract loop body to function?
		
		if backoff_delay > 0 {
			let mut delay = delay_for (Duration::from_millis (backoff_delay)).fuse ();
			
			if futures::select! (
				_ = delay => false,
				_ = shutdown_oneshot => true,
			) {
				info! ("Received graceful shutdown");
				break;
			}
		}
		
		debug! ("http_listen");
		
		let req_req = state.client.get (&format! ("{}/http_listen/{}", state.config.relay_url, config_file.name)).send ();
		
		let err_backoff_delay = std::cmp::min (30_000, backoff_delay * 2 + 500);
		
		let req_req = futures::select! {
			r = req_req.fuse () => r,
			_ = shutdown_oneshot => {
				info! ("Received graceful shutdown");
				break;
			},
		};
		
		let req_resp = match req_req {
			Err (e) => {
				if e.is_timeout () {
					error! ("Client-side timeout. Is an overly-aggressive firewall closing long-lived connections? Is the network flakey?");
				}
				else {
					error! ("Err: {:?}", e);
					if backoff_delay != err_backoff_delay {
						error! ("Non-timeout issue, increasing backoff_delay");
						backoff_delay = err_backoff_delay;
					}
				}
				continue;
			},
			Ok (r) => {
				r
			},
		};
		
		if req_resp.status () == StatusCode::NO_CONTENT {
			debug! ("http_listen long poll timed out on the server, good.");
			continue;
		}
		else if req_resp.status () != StatusCode::OK {
			error! ("{}", req_resp.status ());
			let body = req_resp.bytes ().await.unwrap ();
			let body = String::from_utf8 (body.to_vec ()).unwrap ();
			error! ("{}", body);
			if backoff_delay != err_backoff_delay {
				error! ("Non-timeout issue, increasing backoff_delay");
				backoff_delay = err_backoff_delay;
			}
			continue;
		}
		
		// Unpack the requests, spawn them into new tasks, then loop back
		// around.
		
		if handle_req_resp (&state, req_resp).await.is_err () {
			backoff_delay = err_backoff_delay;
			continue;
		}
		
		if backoff_delay != 0 {
			debug! ("backoff_delay = 0");
			backoff_delay = 0;
		}
	}
	
	info! ("Exiting");
	
	Ok (())
}