diff options
Diffstat (limited to 'kittybox-rs/src/webmentions')
-rw-r--r-- | kittybox-rs/src/webmentions/mod.rs | 75 |
1 files changed, 75 insertions, 0 deletions
diff --git a/kittybox-rs/src/webmentions/mod.rs b/kittybox-rs/src/webmentions/mod.rs new file mode 100644 index 0000000..1c4886e --- /dev/null +++ b/kittybox-rs/src/webmentions/mod.rs @@ -0,0 +1,75 @@ +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, +} + +async fn accept_webmention<Q: JobQueue<Webmention>>( + Form(webmention): Form<Webmention>, + Extension(queue): Extension<Q> +) -> 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) -> axum::Router { + // Automatically spawn a background task to handle webmentions + tokio::task::spawn(supervised_webmentions_task(queue.clone())); + + axum::Router::new() + .route("/.kittybox/webmention", + axum::routing::post(accept_webmention::<Q>) + ) + .layer(Extension(queue)) +} + +async fn supervisor<E: std::error::Error + std::fmt::Debug + Send + 'static, A: futures_util::Future<Output = Result<std::convert::Infallible, E>> + Send + 'static, F: FnMut() -> A>(mut f: F) -> std::convert::Infallible { + loop { + match tokio::task::spawn(f()).await { + Err(e) => tracing::error!("background task exited unexpectedly: {}", e), + Ok(Err(e)) => tracing::error!("background task returned error: {}", e), + Ok(Ok(_)) => unreachable!("task's Ok is Infallible") + } + } +} + +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!() +} + +async fn supervised_webmentions_task<Q: JobQueue<Webmention>>(queue: Q) { + supervisor::<Q::Error, _, _>(|| process_webmentions_from_queue(queue.clone())).await; +} |