caveman/gatherer/src/http_server.rs

use std::{
    collections::{HashMap, HashSet}, net::SocketAddr, ops::Deref
};
use askama::Template;
use axum::{
    async_trait,
    Extension,
    extract::{self, FromRequestParts},
    http::{StatusCode, Request, Response, request::Parts},
    response::IntoResponse,
    routing::get,
    Router,
    middleware::{Next, self},
    body::{Body},
};
use futures::{stream, StreamExt};
use futures::future::{join, join_all};
use cave::{
    firehose::FirehoseFactory,
    store::{Store, TREND_POOL_SIZE}, PERIODS,
    systemd, db::Database,
};

use metrics_exporter_prometheus::PrometheusHandle;
use crate::{
    html_template::HtmlTemplate,
    trends::{TrendAnalyzer, TrendsResults},
};
use tower_http::services::ServeDir;

mod token_donate;
mod token_collect;

type Languages = Vec<String>;

#[derive(Clone)]
pub struct ServerState {
    store: Store,
    db: Database,
    http_client: reqwest::Client,
}

impl ServerState {
    async fn query_trends(&self, language: Option<String>) -> (TrendsResults, Languages, HashMap<String, String>) {
        let mut store = self.store.clone();
        let mut store_ = self.store.clone();

        let (results, mut languages) = join(async move {
            TrendAnalyzer::run(&mut store_, TREND_POOL_SIZE, PERIODS, language)
                .await
                .unwrap()
        }, async {
            store.get_languages()
                .await
                .unwrap()
        }).await;
        languages.sort();

        let tags = results.iter()
            .flat_map(|(_until, _period, result)| result.iter().map(|(_score, tag)| tag.name.to_string()))
            .collect::<HashSet<String>>();
        let tag_images = join_all(tags.into_iter().map(|name| {
            let mut store = store.clone();
            async move {
                let images = store.get_tag_images(&name)
                    .await
                    .unwrap()
                    .into_iter()
                    .enumerate()
                    .flat_map(|(i, url)| if i == 0 {
                        ["".to_owned(), url]
                    } else {
                        [" ".to_owned(), url]
                    })
                    .collect::<String>();
                (name, images)
            }
        })).await.into_iter().collect();

        (results, languages, tag_images)
    }
}

#[async_trait]
impl<S> FromRequestParts<S> for ServerState
where
    S: Send + Sync,
{
    type Rejection = (StatusCode, String);

    async fn from_request_parts<'life0, 'life1>(
        parts: &'life0 mut Parts,
        state: &'life1 S
    ) -> Result<Self, Self::Rejection> {
        let Extension(state) = Extension::<ServerState>::from_request_parts(parts, state)
            .await
            .map_err(internal_error)?;

        Ok(state)
    }
}

/// Utility function for mapping any error into a `500 Internal Server Error`
/// response.
fn internal_error<E>(err: E) -> (StatusCode, String)
where
    E: std::error::Error,
{
    (StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
}


#[derive(Template)]
#[template(path = "trends.html")]
struct TrendsPage {
    language: Option<String>,
    languages: Vec<String>,
    results: TrendsResults,
    tag_images: HashMap<String, String>,
}

impl TrendsPage {
    async fn generate(language: Option<String>, state: ServerState) -> Self {
        let (results, languages, tag_images) = state.query_trends(language.clone()).await;

        // redis queries done, data is ready for rendering, means the
        // service is very much alive:
        systemd::watchdog();

        TrendsPage {
            results,
            language,
            languages,
            tag_images,
        }
    }

    fn template(self) -> HtmlTemplate<Self> {
        HtmlTemplate(self)
    }
}

async fn trends_page(
    state: ServerState,
    language: Option<String>,
) -> Response<Body> {
    let lang = if language.is_some() { "some" } else { "any" };
    let page = TrendsPage::generate(language, state)
        .await;
    let res = page.template().into_response();
    metrics::increment_counter!("trends_page_requests", "lang" => lang);
    res
}

async fn home(Extension(state): Extension<ServerState>) -> Response<Body> {
    trends_page(state, None).await
}

async fn in_language(
    Extension(state): Extension<ServerState>,
    extract::Path(language): extract::Path<String>,
) -> Response<Body> {
    trends_page(state, Some(language)).await
}

async fn streaming_api(
    Extension(firehose_factory): Extension<FirehoseFactory>,
) -> Response<Body> {
    let firehose = firehose_factory.produce()
        .await
        .expect("firehose");
    let stream = stream::once(async { Ok::<Vec<u8>, axum::Error>(b":)\n".to_vec()) })
        .chain(
            firehose.flat_map(|(event_type, data)|
                              stream::iter([
                                  Ok(b"event: ".to_vec()),
                                  Ok(event_type),
                                  Ok(b"\ndata: ".to_vec()),
                                  Ok(data),
                                  Ok(b"\n\n".to_vec()),
                              ].into_iter()).boxed())
        );
    let body = Body::from_stream(stream);

    Response::builder()
        .status(200)
        .header("content-type", "text/event-stream")
        .header("cache-control", "no-store")
        .body(body)
        .expect("Response")
}

async fn print_request(
    req: Request<Body>,
    next: Next,
) -> Result<impl IntoResponse, (StatusCode, String)> {
    tracing::info!(
        "{} {} {:?}",
        req.method(),
        req.uri(),
        req.headers().get("user-agent")
            .and_then(|ua| ua.to_str().ok())
            .unwrap_or("-")
    );
    let res = next.run(req).await;

    Ok(res)
}

pub async fn start(
    listen_port: u16,
    store: Store,
    db: Database,
    http_client: reqwest::Client,
    firehose_factory: FirehoseFactory,
    recorder: PrometheusHandle,
) {
    cave::systemd::status("Starting HTTP server");

    // build our application with some routes
    let app = Router::new()
        .route("/", get(home))
        .route("/in/:language", get(in_language))
        .route("/api/v1/streaming/public", get(streaming_api))
        .route("/token/donate", get(token_donate::get_token_donate).post(token_donate::post_token_donate))
        .route("/token/collect/:host", get(token_collect::get_token_collect))
        .route("/token/thanks", get(token_collect::get_token_thanks))
        .layer(Extension(ServerState { store, db, http_client }))
        .layer(Extension(firehose_factory))
        .route("/metrics", get(|| async move {
            recorder.render().into_response()
        }))
        .layer(middleware::from_fn(print_request))
        .nest_service("/assets", ServeDir::new("assets"));

    // run it
    let addr = SocketAddr::from(([0, 0, 0, 0], listen_port));
    let listener = tokio::net::TcpListener::bind(&addr)
        .await
        .unwrap();
    axum::serve(listener, app.into_make_service())
        .await
        .unwrap();
}
gatherer/assets/hour_graph.js: round curves for relevant period 2022-12-02 22:04:01 +01:00			`use std::{`
delint 2024-03-27 02:11:35 +01:00			`collections::{HashMap, HashSet}, net::SocketAddr, ops::Deref`
gatherer/assets/hour_graph.js: round curves for relevant period 2022-12-02 22:04:01 +01:00			`};`
gather: add webserver 2022-11-06 01:29:58 +01:00			`use askama::Template;`
			`use axum::{`
			`async_trait,`
			`Extension,`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`extract::{self, FromRequestParts},`
			`http::{StatusCode, Request, Response, request::Parts},`
gatherer: delint 2022-11-06 02:20:22 +01:00			`response::IntoResponse,`
gather: add webserver 2022-11-06 01:29:58 +01:00			`routing::get,`
gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`Router,`
			`middleware::{Next, self},`
delint 2024-03-27 02:11:35 +01:00			`body::{Body},`
gather: add webserver 2022-11-06 01:29:58 +01:00			`};`
delint 2024-03-27 02:11:35 +01:00			`use futures::{stream, StreamExt};`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`use futures::future::{join, join_all};`
gatherer: move mod firehose to cave 2022-11-16 19:12:25 +01:00			`use cave::{`
			`firehose::FirehoseFactory,`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`store::{Store, TREND_POOL_SIZE}, PERIODS,`
begin collecting tokens 2023-08-08 18:42:34 +02:00			`systemd, db::Database,`
gatherer: move mod firehose to cave 2022-11-16 19:12:25 +01:00			`};`
delint 2024-03-27 02:11:35 +01:00
gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`use metrics_exporter_prometheus::PrometheusHandle;`
gather: add webserver 2022-11-06 01:29:58 +01:00			`use crate::{`
			`html_template::HtmlTemplate,`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`trends::{TrendAnalyzer, TrendsResults},`
gather: add webserver 2022-11-06 01:29:58 +01:00			`};`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`use tower_http::services::ServeDir;`
gather: add webserver 2022-11-06 01:29:58 +01:00
begin collecting tokens 2023-08-08 18:42:34 +02:00			`mod token_donate;`
			`mod token_collect;`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00
			`type Languages = Vec<String>;`

			`#[derive(Clone)]`
			`pub struct ServerState {`
			`store: Store,`
begin collecting tokens 2023-08-08 18:42:34 +02:00			`db: Database,`
gatherer: use one shared http_client 2023-08-09 00:05:26 +02:00			`http_client: reqwest::Client,`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`}`

			`impl ServerState {`
gatherer/http_server: move metric trends_page_time to TrendAnalyzer.run() for higher granularity 2022-12-30 03:22:35 +01:00			`async fn query_trends(&self, language: Option<String>) -> (TrendsResults, Languages, HashMap<String, String>) {`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`let mut store = self.store.clone();`
			`let mut store_ = self.store.clone();`

			`let (results, mut languages) = join(async move {`
			`TrendAnalyzer::run(&mut store_, TREND_POOL_SIZE, PERIODS, language)`
			`.await`
			`.unwrap()`
			`}, async {`
			`store.get_languages()`
			`.await`
			`.unwrap()`
			`}).await;`
			`languages.sort();`

			`let tags = results.iter()`
			`.flat_map(\|(_until, _period, result)\| result.iter().map(\|(_score, tag)\| tag.name.to_string()))`
			`.collect::<HashSet<String>>();`
			`let tag_images = join_all(tags.into_iter().map(\|name\| {`
			`let mut store = store.clone();`
			`async move {`
			`let images = store.get_tag_images(&name)`
			`.await`
			`.unwrap()`
			`.into_iter()`
			`.enumerate()`
			`.flat_map(\|(i, url)\| if i == 0 {`
			`["".to_owned(), url]`
			`} else {`
			`[" ".to_owned(), url]`
			`})`
			`.collect::<String>();`
			`(name, images)`
			`}`
			`})).await.into_iter().collect();`

			`(results, languages, tag_images)`
			`}`
			`}`
gather: add webserver 2022-11-06 01:29:58 +01:00
			`#[async_trait]`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`impl<S> FromRequestParts<S> for ServerState`
gather: add webserver 2022-11-06 01:29:58 +01:00			`where`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`S: Send + Sync,`
gather: add webserver 2022-11-06 01:29:58 +01:00			`{`
			`type Rejection = (StatusCode, String);`

update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`async fn from_request_parts<'life0, 'life1>(`
			`parts: &'life0 mut Parts,`
			`state: &'life1 S`
			`) -> Result<Self, Self::Rejection> {`
			`let Extension(state) = Extension::<ServerState>::from_request_parts(parts, state)`
gather: add webserver 2022-11-06 01:29:58 +01:00			`.await`
			`.map_err(internal_error)?;`

gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`Ok(state)`
gather: add webserver 2022-11-06 01:29:58 +01:00			`}`
			`}`

			/// Utility function for mapping any error into a `500 Internal Server Error`
			`/// response.`
			`fn internal_error<E>(err: E) -> (StatusCode, String)`
			`where`
			`E: std::error::Error,`
			`{`
			`(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())`
			`}`


			`#[derive(Template)]`
			`#[template(path = "trends.html")]`
			`struct TrendsPage {`
			`language: Option<String>,`
gatherer: show languages 2022-11-10 02:47:09 +01:00			`languages: Vec<String>,`
gather: add webserver 2022-11-06 01:29:58 +01:00			`results: TrendsResults,`
gatherer: show images 2022-11-29 01:51:22 +01:00			`tag_images: HashMap<String, String>,`
gather: add webserver 2022-11-06 01:29:58 +01:00			`}`

			`impl TrendsPage {`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`async fn generate(language: Option<String>, state: ServerState) -> Self {`
gatherer/http_server: move metric trends_page_time to TrendAnalyzer.run() for higher granularity 2022-12-30 03:22:35 +01:00			`let (results, languages, tag_images) = state.query_trends(language.clone()).await;`
gather: add webserver 2022-11-06 01:29:58 +01:00
gatherer: add watchdog triggered by wget 2022-11-14 21:52:48 +01:00			`// redis queries done, data is ready for rendering, means the`
			`// service is very much alive:`
			`systemd::watchdog();`

gather: add webserver 2022-11-06 01:29:58 +01:00			`TrendsPage {`
gatherer: show images 2022-11-29 01:51:22 +01:00			`results,`
gather: add webserver 2022-11-06 01:29:58 +01:00			`language,`
gatherer: show languages 2022-11-10 02:47:09 +01:00			`languages,`
gatherer: show images 2022-11-29 01:51:22 +01:00			`tag_images,`
gather: add webserver 2022-11-06 01:29:58 +01:00			`}`
			`}`

			`fn template(self) -> HtmlTemplate<Self> {`
			`HtmlTemplate(self)`
			`}`
			`}`

gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`async fn trends_page(`
			`state: ServerState,`
			`language: Option<String>,`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`) -> Response<Body> {`
gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`let lang = if language.is_some() { "some" } else { "any" };`
			`let page = TrendsPage::generate(language, state)`
			`.await;`
			`let res = page.template().into_response();`
			`metrics::increment_counter!("trends_page_requests", "lang" => lang);`
			`res`
			`}`

update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`async fn home(Extension(state): Extension<ServerState>) -> Response<Body> {`
gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`trends_page(state, None).await`
gather: add webserver 2022-11-06 01:29:58 +01:00			`}`

			`async fn in_language(`
gatherer/request_mux: eliminate 2022-12-30 01:56:20 +01:00			`Extension(state): Extension<ServerState>,`
gather: add webserver 2022-11-06 01:29:58 +01:00			`extract::Path(language): extract::Path<String>,`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`) -> Response<Body> {`
gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`trends_page(state, Some(language)).await`
gather: add webserver 2022-11-06 01:29:58 +01:00			`}`

gatherer/firehose: add 2022-11-15 00:45:02 +01:00			`async fn streaming_api(`
			`Extension(firehose_factory): Extension<FirehoseFactory>,`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`) -> Response<Body> {`
gatherer/firehose: add 2022-11-15 00:45:02 +01:00			`let firehose = firehose_factory.produce()`
			`.await`
			`.expect("firehose");`
firehose: convey event_type 2023-10-01 23:19:00 +02:00			`let stream = stream::once(async { Ok::<Vec<u8>, axum::Error>(b":)\n".to_vec()) })`
gatherer/firehose: add 2022-11-15 00:45:02 +01:00			`.chain(`
firehose: convey event_type 2023-10-01 23:19:00 +02:00			`firehose.flat_map(\|(event_type, data)\|`
			`stream::iter([`
			`Ok(b"event: ".to_vec()),`
			`Ok(event_type),`
			`Ok(b"\ndata: ".to_vec()),`
			`Ok(data),`
			`Ok(b"\n\n".to_vec()),`
			`].into_iter()).boxed())`
gatherer/firehose: add 2022-11-15 00:45:02 +01:00			`);`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`let body = Body::from_stream(stream);`
gatherer/firehose: add 2022-11-15 00:45:02 +01:00
			`Response::builder()`
			`.status(200)`
			`.header("content-type", "text/event-stream")`
			`.header("cache-control", "no-store")`
delint 2024-03-27 02:11:35 +01:00			`.body(body)`
gatherer/firehose: add 2022-11-15 00:45:02 +01:00			`.expect("Response")`
			`}`

gatherer/http_server: log reqs 2022-11-14 20:54:30 +01:00			`async fn print_request(`
			`req: Request<Body>,`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`next: Next,`
gatherer/http_server: log reqs 2022-11-14 20:54:30 +01:00			`) -> Result<impl IntoResponse, (StatusCode, String)> {`
switch from log to tracing 2022-12-01 01:39:38 +01:00			`tracing::info!(`
gatherer/http_server: log reqs 2022-11-14 20:54:30 +01:00			`"{} {} {:?}",`
			`req.method(),`
			`req.uri(),`
			`req.headers().get("user-agent")`
			`.and_then(\|ua\| ua.to_str().ok())`
			`.unwrap_or("-")`
			`);`
			`let res = next.run(req).await;`

			`Ok(res)`
			`}`

gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`pub async fn start(`
			`listen_port: u16,`
			`store: Store,`
begin collecting tokens 2023-08-08 18:42:34 +02:00			`db: Database,`
gatherer: use one shared http_client 2023-08-09 00:05:26 +02:00			`http_client: reqwest::Client,`
gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`firehose_factory: FirehoseFactory,`
			`recorder: PrometheusHandle,`
			`) {`
gather: add webserver 2022-11-06 01:29:58 +01:00			`cave::systemd::status("Starting HTTP server");`

			`// build our application with some routes`
			`let app = Router::new()`
			`.route("/", get(home))`
			`.route("/in/:language", get(in_language))`
gatherer/firehose: add 2022-11-15 00:45:02 +01:00			`.route("/api/v1/streaming/public", get(streaming_api))`
begin collecting tokens 2023-08-08 18:42:34 +02:00			`.route("/token/donate", get(token_donate::get_token_donate).post(token_donate::post_token_donate))`
			`.route("/token/collect/:host", get(token_collect::get_token_collect))`
			`.route("/token/thanks", get(token_collect::get_token_thanks))`
gatherer: use one shared http_client 2023-08-09 00:05:26 +02:00			`.layer(Extension(ServerState { store, db, http_client }))`
gatherer/firehose: add 2022-11-15 00:45:02 +01:00			`.layer(Extension(firehose_factory))`
gatherer/http_server: add metrics 2022-12-30 02:59:09 +01:00			`.route("/metrics", get(\|\| async move {`
			`recorder.render().into_response()`
			`}))`
gatherer/http_server: log reqs 2022-11-14 20:54:30 +01:00			`.layer(middleware::from_fn(print_request))`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`.nest_service("/assets", ServeDir::new("assets"));`
gather: add webserver 2022-11-06 01:29:58 +01:00
			`// run it`
example vm setup 2023-08-08 22:06:06 +02:00			`let addr = SocketAddr::from(([0, 0, 0, 0], listen_port));`
update axum and reqwest dependencies to http-1.0 2024-03-27 02:03:49 +01:00			`let listener = tokio::net::TcpListener::bind(&addr)`
			`.await`
			`.unwrap();`
			`axum::serve(listener, app.into_make_service())`
gather: add webserver 2022-11-06 01:29:58 +01:00			`.await`
			`.unwrap();`
			`}`