use axum::{Form, response::{IntoResponse, Response}, Extension}; use axum::http::StatusCode; use self::queue::JobQueue; pub mod queue; #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))] pub struct Webmention { source: String, target: String, } impl queue::JobItem for Webmention {} impl queue::PostgresJobItem for Webmention { const DATABASE_NAME: &'static str = "kittybox.incoming_webmention_queue"; const NOTIFICATION_CHANNEL: &'static str = "incoming_webmention"; } async fn accept_webmention<Q: JobQueue<Webmention>>( Extension(queue): Extension<Q>, Form(webmention): Form<Webmention>, ) -> Response { if let Err(err) = webmention.source.parse::<url::Url>() { return (StatusCode::BAD_REQUEST, err.to_string()).into_response() } if let Err(err) = webmention.target.parse::<url::Url>() { return (StatusCode::BAD_REQUEST, err.to_string()).into_response() } match queue.put(&webmention).await { Ok(id) => (StatusCode::ACCEPTED, [ ("Location", format!("/.kittybox/webmention/{id}")) ]).into_response(), Err(err) => (StatusCode::INTERNAL_SERVER_ERROR, [ ("Content-Type", "text/plain") ], err.to_string()).into_response() } } pub fn router<Q: JobQueue<Webmention>>(queue: Q, cancellation_token: tokio_util::sync::CancellationToken) -> (axum::Router, SupervisedTask) { // Automatically spawn a background task to handle webmentions let bgtask_handle = supervised_webmentions_task(queue.clone(), cancellation_token); let router = axum::Router::new() .route("/.kittybox/webmention", axum::routing::post(accept_webmention::<Q>) ) .layer(Extension(queue)); (router, bgtask_handle) } #[derive(thiserror::Error, Debug)] pub enum SupervisorError { #[error("the task was explicitly cancelled")] Cancelled } pub type SupervisedTask = tokio::task::JoinHandle<Result<std::convert::Infallible, SupervisorError>>; pub fn supervisor<E, A, F>(mut f: F, cancellation_token: tokio_util::sync::CancellationToken) -> SupervisedTask where E: std::error::Error + std::fmt::Debug + Send + 'static, A: std::future::Future<Output = Result<std::convert::Infallible, E>> + Send + 'static, F: FnMut() -> A + Send + 'static { let supervisor_future = async move { loop { // Don't spawn the task if we are already cancelled, but // have somehow missed it (probably because the task // crashed and we immediately received a cancellation // request after noticing the crashed task) if cancellation_token.is_cancelled() { return Err(SupervisorError::Cancelled) } let task = tokio::task::spawn(f()); tokio::select! { _ = cancellation_token.cancelled() => { tracing::info!("Shutdown of background task {:?} requested.", std::any::type_name::<A>()); return Err(SupervisorError::Cancelled) } task_result = task => match task_result { Err(e) => tracing::error!("background task {:?} exited unexpectedly: {}", std::any::type_name::<A>(), e), Ok(Err(e)) => tracing::error!("background task {:?} returned error: {}", std::any::type_name::<A>(), e), Ok(Ok(_)) => unreachable!("task's Ok is Infallible") } } tracing::debug!("Sleeping for a little while to back-off..."); tokio::time::sleep(std::time::Duration::from_secs(5)).await; } }; #[cfg(not(tokio_unstable))] return tokio::task::spawn(supervisor_future); #[cfg(tokio_unstable)] return tokio::task::Builder::new() .name(format!("supervisor for background task {}", std::any::type_name::<A>()).as_str()) .spawn(supervisor_future) .unwrap(); } async fn process_webmentions_from_queue<Q: JobQueue<Webmention>>(queue: Q) -> Result<std::convert::Infallible, Q::Error> { use futures_util::StreamExt; use self::queue::Job; let mut stream = queue.into_stream().await?; while let Some(item) = stream.next().await.transpose()? { let job = item.job(); let (source, target) = ( job.source.parse::<url::Url>().unwrap(), job.target.parse::<url::Url>().unwrap() ); todo!() } unreachable!() } fn supervised_webmentions_task<Q: JobQueue<Webmention>>(queue: Q, cancellation_token: tokio_util::sync::CancellationToken) -> SupervisedTask { supervisor::<Q::Error, _, _>(move || process_webmentions_from_queue(queue.clone()), cancellation_token) }