finish

crates/kitsune-activitypub/Cargo.toml

@@ -5,7 +5,6 @@ edition.workspace = true
 version.workspace = true
 
 [dependencies]
-async-recursion = "1.0.5"
 async-trait = "0.1.74"
 autometrics = { version = "0.6.0", default-features = false }
 base64-simd = { version = "0.8.0", features = ["unstable"] }

crates/kitsune-activitypub/src/error.rs

@@ -30,6 +30,15 @@ pub enum Error {
     #[error(transparent)]
     FederationFilter(#[from] kitsune_federation_filter::error::Error),
 
+    #[error(transparent)]
+    FetchAccount(BoxError),
+
+    #[error(transparent)]
+    FetchEmoji(BoxError),
+
+    #[error(transparent)]
+    FetchPost(BoxError),
+
     #[error(transparent)]
     Http(#[from] http::Error),
 

crates/kitsune-activitypub/src/fetcher/mod.rs

@@ -6,7 +6,10 @@ use kitsune_cache::ArcCache;
 use kitsune_core::{
     consts::USER_AGENT,
     error::BoxError,
-    traits::{fetcher::AccountFetchOptions, Fetcher as FetcherTrait, Resolver},
+    traits::{
+        fetcher::{AccountFetchOptions, PostFetchOptions},
+        Fetcher as FetcherTrait, Resolver,
+    },
 };
 use kitsune_db::{
     model::{account::Account, custom_emoji::CustomEmoji, post::Post},
@@ -130,7 +133,7 @@ impl FetcherTrait for Fetcher {
         Ok(self.fetch_emoji(url).await?)
     }
 
-    async fn fetch_post(&self, url: &str) -> Result<Option<Post>, BoxError> {
-        Ok(self.fetch_object(url).await?)
+    async fn fetch_post(&self, opts: PostFetchOptions<'_>) -> Result<Option<Post>, BoxError> {
+        Ok(self.fetch_object(opts.url, opts.call_depth).await?)
     }
 }

crates/kitsune-activitypub/src/fetcher/object.rs

@@ -1,6 +1,5 @@
 use super::Fetcher;
 use crate::{error::Result, process_new_object, ProcessNewObject};
-use async_recursion::async_recursion;
 use autometrics::autometrics;
 use diesel::{ExpressionMethods, OptionalExtension, QueryDsl, SelectableHelper};
 use diesel_async::RunQueryDsl;
@@ -13,12 +12,9 @@ use scoped_futures::ScopedFutureExt;
 pub const MAX_FETCH_DEPTH: u32 = 30;
 
 impl Fetcher {
-    #[async_recursion]
-    pub(crate) async fn fetch_object_inner(
-        &self,
-        url: &str,
-        call_depth: u32,
-    ) -> Result<Option<Post>> {
+    #[instrument(skip(self))]
+    #[autometrics(track_concurrency)]
+    pub(crate) async fn fetch_object(&self, url: &str, call_depth: u32) -> Result<Option<Post>> {
         if call_depth > MAX_FETCH_DEPTH {
             return Ok(None);
         }
@@ -65,10 +61,4 @@ impl Fetcher {
 
         Ok(Some(post))
     }
-
-    #[instrument(skip(self))]
-    #[autometrics(track_concurrency)]
-    pub(crate) async fn fetch_object(&self, url: &str) -> Result<Option<Post>> {
-        self.fetch_object_inner(url, 0).await
-    }
 }

crates/kitsune-activitypub/src/lib.rs

@@ -11,6 +11,7 @@ use diesel_async::{AsyncPgConnection, RunQueryDsl};
 use futures_util::{stream, FutureExt, StreamExt, TryStreamExt};
 use http::Uri;
 use iso8601_timestamp::Timestamp;
+use kitsune_core::traits::{fetcher::PostFetchOptions, Fetcher as FetcherTrait};
 use kitsune_db::{
     model::{
         account::Account,
@@ -79,7 +80,7 @@ async fn handle_mentions(
 async fn handle_custom_emojis(
     db_conn: &mut AsyncPgConnection,
     post_id: Uuid,
-    fetcher: &Fetcher,
+    fetcher: &dyn FetcherTrait,
     tags: &[Tag],
 ) -> Result<()> {
     let emoji_iter = tags.iter().filter(|tag| tag.r#type == TagType::Emoji);
@@ -107,7 +108,8 @@ async fn handle_custom_emojis(
             emoji_text: emoji_tag.name.clone(),
         })
         .try_collect::<Vec<PostCustomEmoji>>()
-        .await?;
+        .await
+        .map_err(Error::FetchEmoji)?;
 
     diesel::insert_into(posts_custom_emojis::table)
         .values(emojis)
@@ -171,7 +173,7 @@ pub struct ProcessNewObject<'a> {
     db_pool: &'a PgPool,
     embed_client: Option<&'a EmbedClient>,
     object: Box<Object>,
-    fetcher: &'a Fetcher,
+    fetcher: &'a dyn FetcherTrait,
     search_backend: &'a AnySearchBackend,
 }
 
@@ -184,7 +186,7 @@ struct PreprocessedObject<'a> {
     content_lang: Language,
     db_pool: &'a PgPool,
     object: Box<Object>,
-    fetcher: &'a Fetcher,
+    fetcher: &'a dyn FetcherTrait,
     search_backend: &'a AnySearchBackend,
 }
 
@@ -208,7 +210,11 @@ async fn preprocess_object(
             return Err(Error::InvalidDocument);
         }
 
-        let Some(author) = fetcher.fetch_actor(attributed_to.into()).await? else {
+        let Some(author) = fetcher
+            .fetch_account(attributed_to.into())
+            .await
+            .map_err(Error::FetchAccount)?
+        else {
             return Err(Error::NotFound);
         };
 
@@ -218,8 +224,14 @@ async fn preprocess_object(
     let visibility = Visibility::from_activitypub(&user, &object).unwrap();
     let in_reply_to_id = if let Some(ref in_reply_to) = object.in_reply_to {
         fetcher
-            .fetch_object_inner(in_reply_to, call_depth + 1)
-            .await?
+            .fetch_post(
+                PostFetchOptions::builder()
+                    .url(in_reply_to)
+                    .call_depth(call_depth + 1)
+                    .build(),
+            )
+            .await
+            .map_err(Error::FetchPost)?
             .map(|post| post.id)
     } else {
         None

crates/kitsune-activitypub/tests/fetcher/basic.rs

@@ -139,7 +139,7 @@ async fn fetch_note() {
             .build();
 
         let note = fetcher
-            .fetch_post("https://corteximplant.com/@0x0/109501674056556919")
+            .fetch_post("https://corteximplant.com/@0x0/109501674056556919".into())
             .await
             .expect("Fetch note")
             .unwrap();

crates/kitsune-activitypub/tests/fetcher/filter.rs

@@ -47,7 +47,7 @@ async fn federation_allow() {
 
         assert!(matches!(
             *fetcher
-                .fetch_post("https://example.com/fakeobject")
+                .fetch_post("https://example.com/fakeobject".into())
                 .await
                 .unwrap_err()
                 .downcast_ref()
@@ -57,7 +57,7 @@ async fn federation_allow() {
 
         assert!(matches!(
             *fetcher
-                .fetch_post("https://other.badstuff.com/otherfake")
+                .fetch_post("https://other.badstuff.com/otherfake".into())
                 .await
                 .unwrap_err()
                 .downcast_ref()
@@ -77,7 +77,7 @@ async fn federation_allow() {
 
         assert!(matches!(
             fetcher
-                .fetch_post("https://corteximplant.com/@0x0/109501674056556919")
+                .fetch_post("https://corteximplant.com/@0x0/109501674056556919".into())
                 .await,
             Ok(..)
         ));
@@ -118,7 +118,7 @@ async fn federation_deny() {
 
         assert!(matches!(
             fetcher
-                .fetch_post("https://example.com/fakeobject")
+                .fetch_post("https://example.com/fakeobject".into())
                 .await
                 .unwrap_err()
                 .downcast_ref()
@@ -127,7 +127,7 @@ async fn federation_deny() {
         ));
         assert!(matches!(
             *fetcher
-                .fetch_post("https://other.badstuff.com/otherfake")
+                .fetch_post("https://other.badstuff.com/otherfake".into())
                 .await
                 .unwrap_err()
                 .downcast_ref()

crates/kitsune-activitypub/tests/fetcher/infinite.rs

@@ -110,7 +110,7 @@ async fn fetch_infinitely_long_reply_chain() {
                 .build();
 
             assert!(fetcher
-                .fetch_post("https://example.com/notes/0")
+                .fetch_post("https://example.com/notes/0".into())
                 .await
                 .is_ok());
         })

crates/kitsune-activitypub/tests/fetcher/origin.rs

@@ -70,7 +70,7 @@ async fn check_ap_id_authority() {
             .build();
 
         let _ = fetcher
-            .fetch_post("https://example.com/@0x0/109501674056556919")
+            .fetch_post("https://example.com/@0x0/109501674056556919".into())
             .await
             .unwrap_err();
     })
@@ -109,7 +109,7 @@ async fn check_ap_content_type() {
 
         assert!(matches!(
             *fetcher
-                .fetch_post("https://corteximplant.com/users/0x0")
+                .fetch_post("https://corteximplant.com/users/0x0".into())
                 .await
                 .unwrap_err()
                 .downcast_ref()

crates/kitsune-core/src/traits/fetcher.rs

@@ -12,15 +12,15 @@ pub struct AccountFetchOptions<'a> {
     #[builder(default, setter(strip_option))]
     pub acct: Option<(&'a str, &'a str)>,
 
-    /// Refetch the ActivityPub entity
+    /// Refetch the account
     ///
     /// This is mainly used to refresh possibly stale actors
     ///
     /// Default: false
     #[builder(default = false)]
     pub refetch: bool,
 
-    /// URL of the ActivityPub entity
+    /// URL of the account
     pub url: &'a str,
 }
 
@@ -30,6 +30,22 @@ impl<'a> From<&'a str> for AccountFetchOptions<'a> {
     }
 }
 
+#[derive(Clone, Copy, Debug, TypedBuilder)]
+pub struct PostFetchOptions<'a> {
+    /// Call depth of recursive calls of the post fetch logic
+    #[builder(default)]
+    pub call_depth: u32,
+
+    /// URL of the object
+    pub url: &'a str,
+}
+
+impl<'a> From<&'a str> for PostFetchOptions<'a> {
+    fn from(value: &'a str) -> Self {
+        Self::builder().url(value).build()
+    }
+}
+
 #[async_trait]
 pub trait Fetcher: Send + Sync + 'static {
     fn resolver(&self) -> Arc<dyn Resolver>;
@@ -41,7 +57,7 @@ pub trait Fetcher: Send + Sync + 'static {
 
     async fn fetch_emoji(&self, url: &str) -> Result<Option<CustomEmoji>, BoxError>;
 
-    async fn fetch_post(&self, url: &str) -> Result<Option<Post>, BoxError>;
+    async fn fetch_post(&self, opts: PostFetchOptions<'_>) -> Result<Option<Post>, BoxError>;
 }
 
 #[async_trait]
@@ -61,8 +77,8 @@ impl Fetcher for Arc<dyn Fetcher> {
         (**self).fetch_emoji(url).await
     }
 
-    async fn fetch_post(&self, url: &str) -> Result<Option<Post>, BoxError> {
-        (**self).fetch_post(url).await
+    async fn fetch_post(&self, opts: PostFetchOptions<'_>) -> Result<Option<Post>, BoxError> {
+        (**self).fetch_post(opts).await
     }
 }
 
@@ -98,9 +114,9 @@ where
         Ok(None)
     }
 
-    async fn fetch_post(&self, url: &str) -> Result<Option<Post>, BoxError> {
+    async fn fetch_post(&self, opts: PostFetchOptions<'_>) -> Result<Option<Post>, BoxError> {
         for fetcher in self {
-            if let Some(post) = fetcher.fetch_post(url).await? {
+            if let Some(post) = fetcher.fetch_post(opts).await? {
                 return Ok(Some(post));
             }
         }

crates/kitsune-email/Cargo.toml

@@ -12,7 +12,6 @@ askama = "0.12.1"
 askama_axum = "0.3.0" # Damn it, cargo. Because "kitsune" uses "askama" with the axum feature, we have to have the crate available here as well..
 diesel = "2.1.4"
 diesel-async = "0.4.1"
-iso8601-timestamp = "0.2.12"
 kitsune-db = { path = "../kitsune-db" }
 kitsune-url = { path = "../kitsune-url" }
 lettre = { version = "0.11.2", default-features = false, features = [

crates/kitsune-email/src/service.rs

@@ -1,8 +1,7 @@
 use crate::{error::Result, mails::confirm_account::ConfirmAccount, MailSender};
-use diesel::{ExpressionMethods, QueryDsl};
+use diesel::{ExpressionMethods, NullableExpressionMethods, QueryDsl};
 use diesel_async::RunQueryDsl;
-use iso8601_timestamp::Timestamp;
-use kitsune_db::{model::user::User, schema::users, PgPool};
+use kitsune_db::{function::now, model::user::User, schema::users, PgPool};
 use kitsune_url::UrlService;
 use lettre::{AsyncSmtpTransport, Tokio1Executor};
 use scoped_futures::ScopedFutureExt;
@@ -22,11 +21,16 @@ impl Mailing {
         self.sender.is_some()
     }
 
+    #[must_use]
+    pub fn sender(&self) -> Option<MailSender<AsyncSmtpTransport<Tokio1Executor>>> {
+        self.sender.clone()
+    }
+
     pub async fn mark_as_confirmed(&self, user_id: Uuid) -> Result<()> {
         self.db_pool
             .with_connection(|db_conn| {
                 diesel::update(users::table.find(user_id))
-                    .set(users::confirmed_at.eq(Timestamp::now_utc()))
+                    .set(users::confirmed_at.eq(now().nullable()))
                     .execute(db_conn)
                     .scoped()
             })
@@ -43,7 +47,7 @@ impl Mailing {
                         .filter(users::confirmation_token.eq(confirmation_token))
                         .filter(users::confirmed_at.is_null()),
                 )
-                .set(users::confirmed_at.eq(Timestamp::now_utc()))
+                .set(users::confirmed_at.eq(now().nullable()))
                 .execute(db_conn)
                 .scoped()
             })