Version 0.5.6

Make response content-type check case insensitive (#111 )
* Make response content-type check case insensitive For wordpress compat * cleaner * clippy * fmt * fmt
2024-06-03 05:49:28 +00:00 · 2024-05-06 11:09:47 +02:00 · 2024-05-06 11:09:23 +02:00 · 2024-05-02 13:06:22 +02:00 · 2024-05-02 06:58:08 -04:00 · 2024-05-02 10:58:56 +02:00
41 changed files with 1404 additions and 563 deletions
--- a/.woodpecker.yml
+++ b/.woodpecker.yml
@ -1,54 +1,56 @@
-pipeline:
+variables:
+  - &rust_image "rust:1.77-bullseye"
+  
+steps:
  cargo_fmt:
    image: rustdocker/rust:nightly
    commands:
      - /root/.cargo/bin/cargo fmt -- --check
-
-  cargo_check:
-    image: rust:1.65-bullseye
-    environment:
-      CARGO_HOME: .cargo
-    commands:
-      - cargo check --all-features --all-targets
+    when:
+      - event: pull_request

  cargo_clippy:
-    image: rust:1.65-bullseye
+    image: *rust_image
    environment:
      CARGO_HOME: .cargo
    commands:
      - rustup component add clippy
-      - cargo clippy --all-targets --all-features --
-        -D warnings -D deprecated -D clippy::perf -D clippy::complexity
-        -D clippy::dbg_macro -D clippy::inefficient_to_string
-        -D clippy::items-after-statements -D clippy::implicit_clone
-        -D clippy::wildcard_imports -D clippy::cast_lossless
-        -D clippy::manual_string_new -D clippy::redundant_closure_for_method_calls
-      - cargo clippy --all-features -- -D clippy::unwrap_used
+      - cargo clippy --all-targets --all-features
+    when:
+      - event: pull_request

  cargo_test:
-    image: rust:1.65-bullseye
+    image: *rust_image
    environment:
      CARGO_HOME: .cargo
    commands:
      - cargo test --all-features --no-fail-fast
+    when:
+      - event: pull_request

  cargo_doc:
-    image: rust:1.65-bullseye
+    image: *rust_image
    environment:
      CARGO_HOME: .cargo
    commands:
      - cargo doc --all-features
+    when:
+      - event: pull_request

  cargo_run_actix_example:
-    image: rust:1.65-bullseye
+    image: *rust_image
    environment:
      CARGO_HOME: .cargo
    commands:
      - cargo run --example local_federation actix-web
+    when:
+      - event: pull_request

  cargo_run_axum_example:
-    image: rust:1.65-bullseye
+    image: *rust_image
    environment:
      CARGO_HOME: .cargo
    commands:
      - cargo run --example local_federation axum
+    when:
+      - event: pull_request
--- a/Cargo.toml
+++ b/Cargo.toml
@ -1,6 +1,6 @@
 [package]
 name = "activitypub_federation"
-version = "0.4.2"
+version = "0.5.6"
 edition = "2021"
 description = "High-level Activitypub framework"
 keywords = ["activitypub", "activitystreams", "federation", "fediverse"]
@ -8,72 +8,94 @@ license = "AGPL-3.0"
 repository = "https://github.com/LemmyNet/activitypub-federation-rust"
 documentation = "https://docs.rs/activitypub_federation/"

+[features]
+default = ["actix-web", "axum"]
+actix-web = ["dep:actix-web"]
+axum = ["dep:axum", "dep:tower", "dep:hyper", "dep:http-body-util"]
+diesel = ["dep:diesel"]
+
+[lints.rust]
+warnings = "deny"
+deprecated = "deny"
+
+[lints.clippy]
+perf = "deny"
+complexity = "deny"
+dbg_macro  = "deny"
+inefficient_to_string = "deny"
+items-after-statements  = "deny" 
+implicit_clone = "deny"
+wildcard_imports = "deny"
+cast_lossless = "deny"
+manual_string_new = "deny"
+redundant_closure_for_method_calls = "deny"
+unwrap_used = "deny"
+
 [dependencies]
-chrono = { version = "0.4.26", features = ["clock"], default-features = false }
-serde = { version = "1.0.164", features = ["derive"] }
-async-trait = "0.1.68"
-url = { version = "2.4.0", features = ["serde"] }
-serde_json = { version = "1.0.96", features = ["preserve_order"] }
-anyhow = "1.0.71"
-reqwest = { version = "0.11.18", features = ["json", "stream"] }
-reqwest-middleware = "0.2.2"
-tracing = "0.1.37"
-base64 = "0.21.2"
-openssl = "0.10.54"
-once_cell = "1.18.0"
-http = "0.2.9"
-sha2 = "0.10.6"
-thiserror = "1.0.40"
-derive_builder = "0.12.0"
-itertools = "0.10.5"
-dyn-clone = "1.0.11"
+chrono = { version = "0.4.38", features = ["clock"], default-features = false }
+serde = { version = "1.0.200", features = ["derive"] }
+async-trait = "0.1.80"
+url = { version = "2.5.0", features = ["serde"] }
+serde_json = { version = "1.0.116", features = ["preserve_order"] }
+reqwest = { version = "0.11.27", features = ["json", "stream"] }
+reqwest-middleware = "0.2.5"
+tracing = "0.1.40"
+base64 = "0.22.1"
+openssl = "0.10.64"
+once_cell = "1.19.0"
+http = "0.2.12"
+sha2 = "0.10.8"
+thiserror = "1.0.59"
+derive_builder = "0.20.0"
+itertools = "0.12.1"
+dyn-clone = "1.0.17"
 enum_delegate = "0.2.0"
-httpdate = "1.0.2"
-http-signature-normalization-reqwest = { version = "0.8.0", default-features = false, features = [
+httpdate = "1.0.3"
+http-signature-normalization-reqwest = { version = "0.10.0", default-features = false, features = [
  "sha-2",
  "middleware",
+  "default-spawner",
 ] }
 http-signature-normalization = "0.7.0"
-bytes = "1.4.0"
-futures-core = { version = "0.3.28", default-features = false }
-pin-project-lite = "0.2.9"
+bytes = "1.6.0"
+futures-core = { version = "0.3.30", default-features = false }
+pin-project-lite = "0.2.14"
 activitystreams-kinds = "0.3.0"
-regex = { version = "1.8.4", default-features = false, features = ["std"] }
-tokio = { version = "1.21.2", features = [
+regex = { version = "1.10.4", default-features = false, features = ["std", "unicode-case"] }
+tokio = { version = "1.37.0", features = [
  "sync",
  "rt",
  "rt-multi-thread",
  "time",
 ] }
+diesel = { version = "2.1.6", features = ["postgres"], default-features = false, optional = true }
+futures = "0.3.30"
+moka = { version = "0.12.7", features = ["future"] }

 # Actix-web
-actix-web = { version = "4.3.1", default-features = false, optional = true }
+actix-web = { version = "4.5.1", default-features = false, optional = true }

 # Axum
-axum = { version = "0.6.18", features = [
+axum = { version = "0.6.20", features = [
  "json",
  "headers",
 ], default-features = false, optional = true }
 tower = { version = "0.4.13", optional = true }
 hyper = { version = "0.14", optional = true }
-displaydoc = "0.2.4"
-
-[features]
-default = ["actix-web", "axum"]
-actix-web = ["dep:actix-web"]
-axum = ["dep:axum", "dep:tower", "dep:hyper"]
+http-body-util = {version = "0.1.1", optional = true }

 [dev-dependencies]
+anyhow = "1.0.82"
 rand = "0.8.5"
-env_logger = "0.10.0"
-tower-http = { version = "0.4.0", features = ["map-request-body", "util"] }
-axum = { version = "0.6.18", features = [
+env_logger = "0.11.3"
+tower-http = { version = "0.5.2", features = ["map-request-body", "util"] }
+axum = { version = "0.6.20", features = [
  "http1",
  "tokio",
  "query",
 ], default-features = false }
-axum-macros = "0.3.7"
-tokio = { version = "1.21.2", features = ["full"] }
+axum-macros = "0.3.8"
+tokio = { version = "1.37.0", features = ["full"] }

 [profile.dev]
 strip = "symbols"
--- a/README.md
+++ b/README.md
@ -2,7 +2,7 @@ Activitypub-Federation
 ===
 [![Crates.io](https://img.shields.io/crates/v/activitypub-federation.svg)](https://crates.io/crates/activitypub-federation)
 [![Documentation](https://shields.io/docsrs/activitypub_federation)](https://docs.rs/activitypub-federation/)
-[![Build Status](https://drone.join-lemmy.org/api/badges/LemmyNet/activitypub-federation-rust/status.svg)](https://drone.join-lemmy.org/LemmyNet/activitypub-federation-rust)
+[![Build Status](https://woodpecker.join-lemmy.org/api/badges/LemmyNet/activitypub-federation-rust/status.svg)](https://drone.join-lemmy.org/LemmyNet/activitypub-federation-rust)

 <!-- be sure to keep this file in sync with docs/01_intro.md -->

--- a/docs/02_overview.md
+++ b/docs/02_overview.md
@ -10,3 +10,12 @@ There are two examples included to see how the library altogether:
 - `live_federation`: A minimal application which can be deployed on a server and federate with other platforms such as Mastodon. For this it needs run at the root of a (sub)domain which is available over HTTPS. Edit `main.rs` to configure the server domain and your Fediverse handle. Once started, it will automatically send a message to you and log any incoming messages.

 To see how this library is used in production, have a look at the [Lemmy federation code](https://github.com/LemmyNet/lemmy/tree/main/crates/apub).
+
+### Security
+This framework does not inherently perform data sanitization upon receiving federated activity data. 
+
+Please, never place implicit trust in the security of data received from the Fediverse. Always keep in mind that malicious entities can be easily created through anonymous fediverse handles.
+
+When implementing our crate in your application, ensure to incorporate data sanitization and validation measures before storing the received data in your database and using it in your user interface. This would significantly reduce the risk of malicious data or actions affecting your application's security and performance.
+
+This framework is designed to simplify your development process, but it's your responsibility to ensure the security of your application. Always follow best practices for data handling, sanitization, and security. 
--- a/docs/03_federating_users.md
+++ b/docs/03_federating_users.md
@ -65,7 +65,7 @@ Besides we also need a second struct to represent the data which gets stored in

 ```rust
 # use url::Url;
-# use chrono::NaiveDateTime;
+# use chrono::{DateTime, Utc};

 pub struct DbUser {
    pub id: i32,
@ -79,7 +79,7 @@ pub struct DbUser {
    pub local: bool,
    pub public_key: String,
    pub private_key: Option<String>,
-    pub last_refreshed_at: NaiveDateTime,
+    pub last_refreshed_at: DateTime<Utc>,
 }
 ```

--- a/docs/05_configuration.md
+++ b/docs/05_configuration.md
@ -14,4 +14,4 @@ let config = FederationConfig::builder()
 # }).unwrap()
 ```

-`debug` is necessary to test federation with http and localhost URLs, but it should never be used in production. The `worker_count` value can be adjusted depending on the instance size. A lower value saves resources on a small instance, while a higher value is necessary on larger instances to keep up with send jobs. `url_verifier` can be used to implement a domain blacklist.
+`debug` is necessary to test federation with http and localhost URLs, but it should never be used in production. `url_verifier` can be used to implement a domain blacklist.
--- a/docs/06_http_endpoints_axum.md
+++ b/docs/06_http_endpoints_axum.md
@ -48,7 +48,7 @@ async fn http_get_user(
 ) -> impl IntoResponse {
    let accept = header_map.get("accept").map(|v| v.to_str().unwrap());
    if accept == Some(FEDERATION_CONTENT_TYPE) {
-        let db_user = data.read_local_user(name).await.unwrap();
+        let db_user = data.read_local_user(&name).await.unwrap();
        let json_user = db_user.into_json(&data).await.unwrap();
        FederationJson(WithContext::new_default(json_user)).into_response()
    }
--- a/docs/07_fetching_data.md
+++ b/docs/07_fetching_data.md
@ -34,7 +34,7 @@ We can similarly dereference a user over webfinger with the following method. It
 # tokio::runtime::Runtime::new().unwrap().block_on(async {
 # let config = FederationConfig::builder().domain("example.com").app_data(db_connection).build().await?;
 # let data = config.to_request_data();
-let user: DbUser = webfinger_resolve_actor("nutomic@lemmy.ml", &data).await?;
+let user: DbUser = webfinger_resolve_actor("ruud@lemmy.world", &data).await?;
 # Ok::<(), anyhow::Error>(())
 # }).unwrap();
 ```
--- a/docs/09_sending_activities.md
+++ b/docs/09_sending_activities.md
@ -4,7 +4,7 @@ To send an activity we need to initialize our previously defined struct, and pic

 ```
 # use activitypub_federation::config::FederationConfig;
-# use activitypub_federation::activity_queue::send_activity;
+# use activitypub_federation::activity_queue::queue_activity;
 # use activitypub_federation::http_signatures::generate_actor_keypair;
 # use activitypub_federation::traits::Actor;
 # use activitypub_federation::fetch::object_id::ObjectId;
@ -25,21 +25,51 @@ let activity = Follow {
    id: "https://lemmy.ml/activities/321".try_into()?
 };
 let inboxes = vec![recipient.shared_inbox_or_inbox()];
-send_activity(activity, &sender, inboxes, &data).await?;
+
+queue_activity(&activity, &sender, inboxes, &data).await?;
 # Ok::<(), anyhow::Error>(())
 # }).unwrap()
 ```

-The list of inboxes gets deduplicated (important for shared inbox). All inboxes on the local
-domain and those which fail the [crate::config::UrlVerifier] check are excluded from delivery.
-For each remaining inbox a background tasks is created. It signs the HTTP header with the given
-private key. Finally the activity is delivered to the inbox.
+The list of inboxes gets deduplicated (important for shared inbox). All inboxes on the local domain and those which fail the [crate::config::UrlVerifier] check are excluded from delivery. For each remaining inbox a background tasks is created. It signs the HTTP header with the given private key. Finally the activity is delivered to the inbox.
+
+It is possible that delivery fails because the target instance is temporarily unreachable. In this case the task is scheduled for retry after a certain waiting time. For each task delivery is retried up to 3 times after the initial attempt. The retry intervals are as follows:

-It is possible that delivery fails because the target instance is temporarily unreachable. In
-this case the task is scheduled for retry after a certain waiting time. For each task delivery
-is retried up to 3 times after the initial attempt. The retry intervals are as follows:
 - one minute, in case of service restart
 - one hour, in case of instance maintenance
 - 2.5 days, in case of major incident with rebuild from backup

 In case [crate::config::FederationConfigBuilder::debug] is enabled, no background thread is used but activities are sent directly on the foreground. This makes it easier to catch delivery errors and avoids complicated steps to await delivery in tests.
+
+In some cases you may want to bypass the builtin activity queue, and implement your own. For example to specify different retry intervals, or to persist retries across application restarts. You can do it with the following code:
+```rust
+# use activitypub_federation::config::FederationConfig;
+# use activitypub_federation::activity_sending::SendActivityTask;
+# use activitypub_federation::http_signatures::generate_actor_keypair;
+# use activitypub_federation::traits::Actor;
+# use activitypub_federation::fetch::object_id::ObjectId;
+# use activitypub_federation::traits::tests::{DB_USER, DbConnection, Follow};
+# tokio::runtime::Runtime::new().unwrap().block_on(async {
+# let db_connection = DbConnection;
+# let config = FederationConfig::builder()
+#     .domain("example.com")
+#     .app_data(db_connection)
+#     .build().await?;
+# let data = config.to_request_data();
+# let sender = DB_USER.clone();
+# let recipient = DB_USER.clone();
+let activity = Follow {
+    actor: ObjectId::parse("https://lemmy.ml/u/nutomic")?,
+    object: recipient.federation_id.clone().into(),
+    kind: Default::default(),
+    id: "https://lemmy.ml/activities/321".try_into()?
+};
+let inboxes = vec![recipient.shared_inbox_or_inbox()];
+
+let sends = SendActivityTask::prepare(&activity, &sender, inboxes, &data).await?;
+for send in sends {
+send.sign_and_send(&data).await?;
+}
+# Ok::<(), anyhow::Error>(())
+# }).unwrap()
+```
--- a/docs/10_fetching_objects_with_unknown_type.md
+++ b/docs/10_fetching_objects_with_unknown_type.md
@ -13,12 +13,13 @@ It is sometimes necessary to fetch from a URL, but we don't know the exact type
 # use url::Url;
 # use activitypub_federation::traits::tests::{Person, Note};

+#[derive(Debug)]
 pub enum SearchableDbObjects {
    User(DbUser),
    Post(DbPost)
 }

-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(untagged)]
 pub enum SearchableObjects {
    Person(Person),
--- a/examples/live_federation/activities/create_post.rs
+++ b/examples/live_federation/activities/create_post.rs
@ -6,7 +6,7 @@ use crate::{
    DbPost,
 };
 use activitypub_federation::{
-    activity_queue::send_activity,
+    activity_sending::SendActivityTask,
    config::Data,
    fetch::object_id::ObjectId,
    kinds::activity::CreateType,
@ -39,7 +39,12 @@ impl CreatePost {
            id: generate_object_id(data.domain())?,
        };
        let create_with_context = WithContext::new_default(create);
-        send_activity(create_with_context, &data.local_user(), vec![inbox], data).await?;
+        let sends =
+            SendActivityTask::prepare(&create_with_context, &data.local_user(), vec![inbox], data)
+                .await?;
+        for send in sends {
+            send.sign_and_send(data).await?;
+        }
        Ok(())
    }
 }
--- a/examples/live_federation/http.rs
+++ b/examples/live_federation/http.rs
@ -61,7 +61,7 @@ pub async fn webfinger(
    data: Data<DatabaseHandle>,
 ) -> Result<Json<Webfinger>, Error> {
    let name = extract_webfinger_name(&query.resource, &data)?;
-    let db_user = data.read_user(&name)?;
+    let db_user = data.read_user(name)?;
    Ok(Json(build_webfinger_response(
        query.resource,
        db_user.ap_id.into_inner(),
--- a/examples/live_federation/main.rs
+++ b/examples/live_federation/main.rs
@ -1,3 +1,5 @@
+#![allow(clippy::unwrap_used)]
+
 use crate::{
    database::Database,
    http::{http_get_user, http_post_user_inbox, webfinger},
--- a/examples/live_federation/objects/person.rs
+++ b/examples/live_federation/objects/person.rs
@ -7,7 +7,7 @@ use activitypub_federation::{
    protocol::{public_key::PublicKey, verification::verify_domains_match},
    traits::{ActivityHandler, Actor, Object},
 };
-use chrono::{Local, NaiveDateTime};
+use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 use url::Url;
@ -21,7 +21,7 @@ pub struct DbUser {
    pub public_key: String,
    // exists only for local users
    pub private_key: Option<String>,
-    last_refreshed_at: NaiveDateTime,
+    last_refreshed_at: DateTime<Utc>,
    pub followers: Vec<Url>,
    pub local: bool,
 }
@ -45,7 +45,7 @@ impl DbUser {
            inbox,
            public_key: keypair.public_key,
            private_key: Some(keypair.private_key),
-            last_refreshed_at: Local::now().naive_local(),
+            last_refreshed_at: Utc::now(),
            followers: vec![],
            local: true,
        })
@ -69,7 +69,7 @@ impl Object for DbUser {
    type Kind = Person;
    type Error = Error;

-    fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    fn last_refreshed_at(&self) -> Option<DateTime<Utc>> {
        Some(self.last_refreshed_at)
    }

@ -114,7 +114,7 @@ impl Object for DbUser {
            inbox: json.inbox,
            public_key: json.public_key.public_key_pem,
            private_key: None,
-            last_refreshed_at: Local::now().naive_local(),
+            last_refreshed_at: Utc::now(),
            followers: vec![],
            local: false,
        })
--- a/examples/live_federation/objects/post.rs
+++ b/examples/live_federation/objects/post.rs
@ -21,7 +21,6 @@ pub struct DbPost {
    pub text: String,
    pub ap_id: ObjectId<DbPost>,
    pub creator: ObjectId<DbUser>,
-    pub local: bool,
 }

 #[derive(Deserialize, Serialize, Debug)]
@ -59,7 +58,15 @@ impl Object for DbPost {
    }

    async fn into_json(self, _data: &Data<Self::DataType>) -> Result<Self::Kind, Self::Error> {
-        unimplemented!()
+        Ok(Note {
+            kind: NoteType::Note,
+            id: self.ap_id,
+            content: self.text,
+            attributed_to: self.creator,
+            to: vec![public()],
+            tag: vec![],
+            in_reply_to: None,
+        })
    }

    async fn verify(
@ -81,7 +88,6 @@ impl Object for DbPost {
            text: json.content,
            ap_id: json.id.clone(),
            creator: json.attributed_to.clone(),
-            local: false,
        };

        let mention = Mention {
--- a/examples/local_federation/activities/follow.rs
+++ b/examples/local_federation/activities/follow.rs
@ -67,7 +67,7 @@ impl ActivityHandler for Follow {
        let id = generate_object_id(data.domain())?;
        let accept = Accept::new(local_user.ap_id.clone(), self, id.clone());
        local_user
-            .send(accept, vec![follower.shared_inbox_or_inbox()], data)
+            .send(accept, vec![follower.shared_inbox_or_inbox()], false, data)
            .await?;
        Ok(())
    }
--- a/examples/local_federation/actix_web/http.rs
+++ b/examples/local_federation/actix_web/http.rs
@ -89,7 +89,7 @@ pub async fn webfinger(
    data: Data<DatabaseHandle>,
 ) -> Result<HttpResponse, Error> {
    let name = extract_webfinger_name(&query.resource, &data)?;
-    let db_user = data.read_user(&name)?;
+    let db_user = data.read_user(name)?;
    Ok(HttpResponse::Ok().json(build_webfinger_response(
        query.resource.clone(),
        db_user.ap_id.into_inner(),
--- a/examples/local_federation/axum/http.rs
+++ b/examples/local_federation/axum/http.rs
@ -78,7 +78,7 @@ async fn webfinger(
    data: Data<DatabaseHandle>,
 ) -> Result<Json<Webfinger>, Error> {
    let name = extract_webfinger_name(&query.resource, &data)?;
-    let db_user = data.read_user(&name)?;
+    let db_user = data.read_user(name)?;
    Ok(Json(build_webfinger_response(
        query.resource,
        db_user.ap_id.into_inner(),
--- a/examples/local_federation/instance.rs
+++ b/examples/local_federation/instance.rs
@ -28,6 +28,7 @@ pub async fn new_instance(
        .domain(hostname)
        .signed_fetch_actor(&system_user)
        .app_data(database)
+        .url_verifier(Box::new(MyUrlVerifier()))
        .debug(true)
        .build()
        .await?;
@ -49,9 +50,11 @@ struct MyUrlVerifier();

 #[async_trait]
 impl UrlVerifier for MyUrlVerifier {
-    async fn verify(&self, url: &Url) -> Result<(), &'static str> {
+    async fn verify(&self, url: &Url) -> Result<(), activitypub_federation::error::Error> {
        if url.domain() == Some("malicious.com") {
-            Err("malicious domain")
+            Err(activitypub_federation::error::Error::Other(
+                "malicious domain".into(),
+            ))
        } else {
            Ok(())
        }
--- a/examples/local_federation/main.rs
+++ b/examples/local_federation/main.rs
@ -1,3 +1,5 @@
+#![allow(clippy::unwrap_used)]
+
 use crate::{
    instance::{listen, new_instance, Webserver},
    objects::post::DbPost,
--- a/examples/local_federation/objects/person.rs
+++ b/examples/local_federation/objects/person.rs
@ -6,7 +6,8 @@ use crate::{
    utils::generate_object_id,
 };
 use activitypub_federation::{
-    activity_queue::send_activity,
+    activity_queue::queue_activity,
+    activity_sending::SendActivityTask,
    config::Data,
    fetch::{object_id::ObjectId, webfinger::webfinger_resolve_actor},
    http_signatures::generate_actor_keypair,
@ -14,7 +15,7 @@ use activitypub_federation::{
    protocol::{context::WithContext, public_key::PublicKey, verification::verify_domains_match},
    traits::{ActivityHandler, Actor, Object},
 };
-use chrono::{Local, NaiveDateTime};
+use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 use url::Url;
@ -28,7 +29,7 @@ pub struct DbUser {
    public_key: String,
    // exists only for local users
    private_key: Option<String>,
-    last_refreshed_at: NaiveDateTime,
+    last_refreshed_at: DateTime<Utc>,
    pub followers: Vec<Url>,
    pub local: bool,
 }
@ -54,7 +55,7 @@ impl DbUser {
            inbox,
            public_key: keypair.public_key,
            private_key: Some(keypair.private_key),
-            last_refreshed_at: Local::now().naive_local(),
+            last_refreshed_at: Utc::now(),
            followers: vec![],
            local: true,
        })
@ -85,7 +86,7 @@ impl DbUser {
        let other: DbUser = webfinger_resolve_actor(other, data).await?;
        let id = generate_object_id(data.domain())?;
        let follow = Follow::new(self.ap_id.clone(), other.ap_id.clone(), id.clone());
-        self.send(follow, vec![other.shared_inbox_or_inbox()], data)
+        self.send(follow, vec![other.shared_inbox_or_inbox()], false, data)
            .await?;
        Ok(())
    }
@ -98,7 +99,7 @@ impl DbUser {
            let user: DbUser = ObjectId::from(f).dereference(data).await?;
            inboxes.push(user.shared_inbox_or_inbox());
        }
-        self.send(create, inboxes, data).await?;
+        self.send(create, inboxes, true, data).await?;
        Ok(())
    }

@ -106,14 +107,23 @@ impl DbUser {
        &self,
        activity: Activity,
        recipients: Vec<Url>,
+        use_queue: bool,
        data: &Data<DatabaseHandle>,
-    ) -> Result<(), <Activity as ActivityHandler>::Error>
+    ) -> Result<(), Error>
    where
        Activity: ActivityHandler + Serialize + Debug + Send + Sync,
        <Activity as ActivityHandler>::Error: From<anyhow::Error> + From<serde_json::Error>,
    {
        let activity = WithContext::new_default(activity);
-        send_activity(activity, self, recipients, data).await?;
+        // Send through queue in some cases and bypass it in others to test both code paths
+        if use_queue {
+            queue_activity(&activity, self, recipients, data).await?;
+        } else {
+            let sends = SendActivityTask::prepare(&activity, self, recipients, data).await?;
+            for send in sends {
+                send.sign_and_send(data).await?;
+            }
+        }
        Ok(())
    }
 }
@ -124,7 +134,7 @@ impl Object for DbUser {
    type Kind = Person;
    type Error = Error;

-    fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    fn last_refreshed_at(&self) -> Option<DateTime<Utc>> {
        Some(self.last_refreshed_at)
    }

@ -166,7 +176,7 @@ impl Object for DbUser {
            inbox: json.inbox,
            public_key: json.public_key.public_key_pem,
            private_key: None,
-            last_refreshed_at: Local::now().naive_local(),
+            last_refreshed_at: Utc::now(),
            followers: vec![],
            local: false,
        };
--- a/src/activity_queue.rs
+++ b/src/activity_queue.rs
@ -3,22 +3,14 @@
 #![doc = include_str!("../docs/09_sending_activities.md")]

 use crate::{
+    activity_sending::{build_tasks, SendActivityTask},
    config::Data,
    error::Error,
-    http_signatures::sign_request,
-    reqwest_shim::ResponseExt,
    traits::{ActivityHandler, Actor},
-    FEDERATION_CONTENT_TYPE,
 };
-use anyhow::anyhow;

-use bytes::Bytes;
 use futures_core::Future;
-use http::{header::HeaderName, HeaderMap, HeaderValue};
-use httpdate::fmt_http_date;
-use itertools::Itertools;
-use openssl::pkey::{PKey, Private};
-use reqwest::Request;
+
 use reqwest_middleware::ClientWithMiddleware;
 use serde::Serialize;
 use std::{
@ -27,16 +19,17 @@ use std::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
-    time::{Duration, SystemTime},
+    time::Duration,
 };
 use tokio::{
    sync::mpsc::{unbounded_channel, UnboundedSender},
    task::{JoinHandle, JoinSet},
 };
-use tracing::{debug, info, warn};
+use tracing::{info, warn};
 use url::Url;

-/// Send a new activity to the given inboxes
+/// Send a new activity to the given inboxes with automatic retry on failure. Alternatively you
+/// can implement your own queue and then send activities using [[crate::activity_sending::SendActivityTask]].
 ///
 /// - `activity`: The activity to be sent, gets converted to json
 /// - `private_key`: Private key belonging to the actor who sends the activity, for signing HTTP
@ -44,63 +37,25 @@ use url::Url;
 /// - `inboxes`: List of remote actor inboxes that should receive the activity. Ignores local actor
 ///              inboxes. Should be built by calling [crate::traits::Actor::shared_inbox_or_inbox]
 ///              for each target actor.
-pub async fn send_activity<Activity, Datatype, ActorType>(
-    activity: Activity,
+pub async fn queue_activity<Activity, Datatype, ActorType>(
+    activity: &Activity,
    actor: &ActorType,
    inboxes: Vec<Url>,
    data: &Data<Datatype>,
-) -> Result<(), <Activity as ActivityHandler>::Error>
+) -> Result<(), Error>
 where
-    Activity: ActivityHandler + Serialize,
-    <Activity as ActivityHandler>::Error: From<anyhow::Error> + From<serde_json::Error>,
+    Activity: ActivityHandler + Serialize + Debug,
    Datatype: Clone,
    ActorType: Actor,
 {
    let config = &data.config;
-    let actor_id = activity.actor();
-    let activity_id = activity.id();
-    let activity_serialized: Bytes = serde_json::to_vec(&activity)?.into();
-    let private_key_pem = actor
-        .private_key_pem()
-        .ok_or_else(|| anyhow!("Actor {actor_id} does not contain a private key for signing"))?;
-
-    // This is a mostly expensive blocking call, we don't want to tie up other tasks while this is happening
-    let private_key = tokio::task::spawn_blocking(move || {
-        PKey::private_key_from_pem(private_key_pem.as_bytes())
-            .map_err(|err| anyhow!("Could not create private key from PEM data:{err}"))
-    })
-    .await
-    .map_err(|err| anyhow!("Error joining:{err}"))??;
-
-    let inboxes: Vec<Url> = inboxes
-        .into_iter()
-        .unique()
-        .filter(|i| !config.is_local_url(i))
-        .collect();
-
-    // This field is only optional to make builder work, its always present at this point
-    let activity_queue = config
-        .activity_queue
-        .as_ref()
-        .expect("Config has activity queue");
-    for inbox in inboxes {
-        if config.verify_url_valid(&inbox).await.is_err() {
-            continue;
-        }
-
-        let message = SendActivityTask {
-            actor_id: actor_id.clone(),
-            activity_id: activity_id.clone(),
-            inbox,
-            activity: activity_serialized.clone(),
-            private_key: private_key.clone(),
-            http_signature_compat: config.http_signature_compat,
-        };
+    let tasks = build_tasks(activity, actor, inboxes, data).await?;

+    for task in tasks {
        // Don't use the activity queue if this is in debug mode, send and wait directly
        if config.debug {
            if let Err(err) = sign_and_send(
-                &message,
+                &task,
                &config.client,
                config.request_timeout,
                Default::default(),
@ -110,145 +65,38 @@ where
                warn!("{err}");
            }
        } else {
-            activity_queue.queue(message).await?;
+            // This field is only optional to make builder work, its always present at this point
+            let activity_queue = config
+                .activity_queue
+                .as_ref()
+                .expect("Config has activity queue");
+            activity_queue.queue(task).await?;
            let stats = activity_queue.get_stats();
            let running = stats.running.load(Ordering::Relaxed);
-            let stats_fmt = format!(
-            "Activity queue stats: pending: {}, running: {}, retries: {}, dead: {}, complete: {}",
-            stats.pending.load(Ordering::Relaxed),
-            running,
-            stats.retries.load(Ordering::Relaxed),
-            stats.dead_last_hour.load(Ordering::Relaxed),
-            stats.completed_last_hour.load(Ordering::Relaxed),
-        );
-            if running == config.worker_count {
-                warn!("Reached max number of send activity workers ({}). Consider increasing worker count to avoid federation delays", config.worker_count);
-                warn!(stats_fmt);
+            if running == config.queue_worker_count && config.queue_worker_count != 0 {
+                warn!("Reached max number of send activity workers ({}). Consider increasing worker count to avoid federation delays", config.queue_worker_count);
+                warn!("{:?}", stats);
            } else {
-                info!(stats_fmt);
+                info!("{:?}", stats);
            }
        }
    }
-
    Ok(())
 }

-#[derive(Clone, Debug)]
-struct SendActivityTask {
-    actor_id: Url,
-    activity_id: Url,
-    activity: Bytes,
-    inbox: Url,
-    private_key: PKey<Private>,
-    http_signature_compat: bool,
-}
-
 async fn sign_and_send(
    task: &SendActivityTask,
    client: &ClientWithMiddleware,
    timeout: Duration,
    retry_strategy: RetryStrategy,
-) -> Result<(), anyhow::Error> {
-    debug!(
-        "Sending {} to {}, contents:\n {}",
-        task.activity_id,
-        task.inbox,
-        serde_json::from_slice::<serde_json::Value>(&task.activity)?
-    );
-    let request_builder = client
-        .post(task.inbox.to_string())
-        .timeout(timeout)
-        .headers(generate_request_headers(&task.inbox));
-    let request = sign_request(
-        request_builder,
-        &task.actor_id,
-        task.activity.clone(),
-        task.private_key.clone(),
-        task.http_signature_compat,
-    )
-    .await?;
-
+) -> Result<(), Error> {
    retry(
-        || {
-            send(
-                task,
-                client,
-                request
-                    .try_clone()
-                    .expect("The body of the request is not cloneable"),
-            )
-        },
+        || task.sign_and_send_internal(client, timeout),
        retry_strategy,
    )
    .await
 }

-async fn send(
-    task: &SendActivityTask,
-    client: &ClientWithMiddleware,
-    request: Request,
-) -> Result<(), anyhow::Error> {
-    let response = client.execute(request).await;
-
-    match response {
-        Ok(o) if o.status().is_success() => {
-            debug!(
-                "Activity {} delivered successfully to {}",
-                task.activity_id, task.inbox
-            );
-            Ok(())
-        }
-        Ok(o) if o.status().is_client_error() => {
-            let text = o.text_limited().await.map_err(Error::other)?;
-            debug!(
-                "Activity {} was rejected by {}, aborting: {}",
-                task.activity_id, task.inbox, text,
-            );
-            Ok(())
-        }
-        Ok(o) => {
-            let status = o.status();
-            let text = o.text_limited().await.map_err(Error::other)?;
-            Err(anyhow!(
-                "Queueing activity {} to {} for retry after failure with status {}: {}",
-                task.activity_id,
-                task.inbox,
-                status,
-                text,
-            ))
-        }
-        Err(e) => {
-            debug!(
-                "Unable to connect to {}, aborting task {}: {}",
-                task.inbox, task.activity_id, e
-            );
-            Ok(())
-        }
-    }
-}
-
-pub(crate) fn generate_request_headers(inbox_url: &Url) -> HeaderMap {
-    let mut host = inbox_url.domain().expect("read inbox domain").to_string();
-    if let Some(port) = inbox_url.port() {
-        host = format!("{}:{}", host, port);
-    }
-
-    let mut headers = HeaderMap::new();
-    headers.insert(
-        HeaderName::from_static("content-type"),
-        HeaderValue::from_static(FEDERATION_CONTENT_TYPE),
-    );
-    headers.insert(
-        HeaderName::from_static("host"),
-        HeaderValue::from_str(&host).expect("Hostname is valid"),
-    );
-    headers.insert(
-        "date",
-        HeaderValue::from_str(&fmt_http_date(SystemTime::now())).expect("Date is valid"),
-    );
-    headers
-}
-
 /// A simple activity queue which spawns tokio workers to send out requests
 /// When creating a queue, it will spawn a task per worker thread
 /// Uses an unbounded mpsc queue for communication (i.e, all messages are in memory)
@ -264,7 +112,7 @@ pub(crate) struct ActivityQueue {
 /// This is a lock-free way to share things between tasks
 /// When reading these values it's possible (but extremely unlikely) to get stale data if a worker task is in the middle of transitioning
 #[derive(Default)]
-struct Stats {
+pub(crate) struct Stats {
    pending: AtomicUsize,
    running: AtomicUsize,
    retries: AtomicUsize,
@ -272,6 +120,20 @@ struct Stats {
    completed_last_hour: AtomicUsize,
 }

+impl Debug for Stats {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Activity queue stats: pending: {}, running: {}, retries: {}, dead: {}, complete: {}",
+            self.pending.load(Ordering::Relaxed),
+            self.running.load(Ordering::Relaxed),
+            self.retries.load(Ordering::Relaxed),
+            self.dead_last_hour.load(Ordering::Relaxed),
+            self.completed_last_hour.load(Ordering::Relaxed)
+        )
+    }
+}
+
 #[derive(Clone, Copy, Default)]
 struct RetryStrategy {
    /// Amount of time in seconds to back off
@ -410,18 +272,25 @@ impl ActivityQueue {
            let mut join_set = JoinSet::new();

            while let Some(message) = retry_receiver.recv().await {
-                // If we're over the limit of retries, wait for them to finish before spawning
-                while join_set.len() >= retry_count {
-                    join_set.join_next().await;
-                }
-
-                join_set.spawn(retry_worker(
+                let retry_task = retry_worker(
                    retry_client.clone(),
                    timeout,
                    message,
                    retry_stats.clone(),
                    retry_strategy,
-                ));
+                );
+
+                if retry_count > 0 {
+                    // If we're over the limit of retries, wait for them to finish before spawning
+                    while join_set.len() >= retry_count {
+                        join_set.join_next().await;
+                    }
+
+                    join_set.spawn(retry_task);
+                } else {
+                    // If the retry worker count is `0` then just spawn and don't use the join_set
+                    tokio::spawn(retry_task);
+                }
            }

            while !join_set.is_empty() {
@ -437,19 +306,26 @@ impl ActivityQueue {
            let mut join_set = JoinSet::new();

            while let Some(message) = receiver.recv().await {
-                // If we're over the limit of workers, wait for them to finish before spawning
-                while join_set.len() >= worker_count {
-                    join_set.join_next().await;
-                }
-
-                join_set.spawn(worker(
+                let task = worker(
                    client.clone(),
                    timeout,
                    message,
                    retry_sender.clone(),
                    sender_stats.clone(),
                    strategy,
-                ));
+                );
+
+                if worker_count > 0 {
+                    // If we're over the limit of workers, wait for them to finish before spawning
+                    while join_set.len() >= worker_count {
+                        join_set.join_next().await;
+                    }
+
+                    join_set.spawn(task);
+                } else {
+                    // If the worker count is `0` then just spawn and don't use the join_set
+                    tokio::spawn(task);
+                }
            }

            drop(retry_sender);
@ -467,9 +343,11 @@ impl ActivityQueue {
        }
    }

-    async fn queue(&self, message: SendActivityTask) -> Result<(), anyhow::Error> {
+    async fn queue(&self, message: SendActivityTask) -> Result<(), Error> {
        self.stats.pending.fetch_add(1, Ordering::Relaxed);
-        self.sender.send(message)?;
+        self.sender
+            .send(message)
+            .map_err(|e| Error::ActivityQueueError(e.0.activity_id))?;

        Ok(())
    }
@ -480,7 +358,7 @@ impl ActivityQueue {

    #[allow(unused)]
    // Drops all the senders and shuts down the workers
-    async fn shutdown(self, wait_for_retries: bool) -> Result<Arc<Stats>, anyhow::Error> {
+    pub(crate) async fn shutdown(self, wait_for_retries: bool) -> Result<Arc<Stats>, Error> {
        drop(self.sender);

        self.sender_task.await?;
@ -501,11 +379,6 @@ pub(crate) fn create_activity_queue(
    retry_count: usize,
    request_timeout: Duration,
 ) -> ActivityQueue {
-    assert!(
-        worker_count > 0,
-        "worker count needs to be greater than zero"
-    );
-
    ActivityQueue::new(client, worker_count, retry_count, request_timeout, 60)
 }

@ -543,17 +416,16 @@ async fn retry<T, E: Display + Debug, F: Future<Output = Result<T, E>>, A: FnMut
 }

 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod tests {
+    use super::*;
+    use crate::http_signatures::generate_actor_keypair;
    use axum::extract::State;
    use bytes::Bytes;
-    use http::StatusCode;
+    use http::{HeaderMap, StatusCode};
    use std::time::Instant;
+    use tracing::debug;

-    use crate::http_signatures::generate_actor_keypair;
-
-    use super::*;
-
-    #[allow(unused)]
    // This will periodically send back internal errors to test the retry
    async fn dodgy_handler(
        State(state): State<Arc<AtomicUsize>>,
@ -579,7 +451,7 @@ mod tests {
            .route("/", post(dodgy_handler))
            .with_state(state);

-        axum::Server::bind(&"0.0.0.0:8001".parse().unwrap())
+        axum::Server::bind(&"0.0.0.0:8002".parse().unwrap())
            .serve(app.into_make_service())
            .await
            .unwrap();
@ -616,10 +488,10 @@ mod tests {
        let keypair = generate_actor_keypair().unwrap();

        let message = SendActivityTask {
-            actor_id: "http://localhost:8001".parse().unwrap(),
-            activity_id: "http://localhost:8001/activity".parse().unwrap(),
+            actor_id: "http://localhost:8002".parse().unwrap(),
+            activity_id: "http://localhost:8002/activity".parse().unwrap(),
            activity: "{}".into(),
-            inbox: "http://localhost:8001".parse().unwrap(),
+            inbox: "http://localhost:8002".parse().unwrap(),
            private_key: keypair.private_key().unwrap(),
            http_signature_compat: true,
        };
--- a/src/activity_sending.rs
+++ b/src/activity_sending.rs
@ -0,0 +1,348 @@
+//! Queue for signing and sending outgoing activities with retry
+//!
+#![doc = include_str!("../docs/09_sending_activities.md")]
+
+use crate::{
+    config::Data,
+    error::Error,
+    http_signatures::sign_request,
+    reqwest_shim::ResponseExt,
+    traits::{ActivityHandler, Actor},
+    FEDERATION_CONTENT_TYPE,
+};
+use bytes::Bytes;
+use futures::StreamExt;
+use http::StatusCode;
+use httpdate::fmt_http_date;
+use itertools::Itertools;
+use openssl::pkey::{PKey, Private};
+use reqwest::{
+    header::{HeaderMap, HeaderName, HeaderValue},
+    Response,
+};
+use reqwest_middleware::ClientWithMiddleware;
+use serde::Serialize;
+use std::{
+    fmt::{Debug, Display},
+    time::{Duration, SystemTime},
+};
+use tracing::debug;
+use url::Url;
+
+#[derive(Clone, Debug)]
+/// All info needed to sign and send one activity to one inbox. You should generally use
+/// [[crate::activity_queue::queue_activity]] unless you want implement your own queue.
+pub struct SendActivityTask {
+    pub(crate) actor_id: Url,
+    pub(crate) activity_id: Url,
+    pub(crate) activity: Bytes,
+    pub(crate) inbox: Url,
+    pub(crate) private_key: PKey<Private>,
+    pub(crate) http_signature_compat: bool,
+}
+
+impl Display for SendActivityTask {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{} to {}", self.activity_id, self.inbox)
+    }
+}
+
+impl SendActivityTask {
+    /// Prepare an activity for sending
+    ///
+    /// - `activity`: The activity to be sent, gets converted to json
+    /// - `inboxes`: List of remote actor inboxes that should receive the activity. Ignores local actor
+    ///              inboxes. Should be built by calling [crate::traits::Actor::shared_inbox_or_inbox]
+    ///              for each target actor.
+    pub async fn prepare<Activity, Datatype, ActorType>(
+        activity: &Activity,
+        actor: &ActorType,
+        inboxes: Vec<Url>,
+        data: &Data<Datatype>,
+    ) -> Result<Vec<SendActivityTask>, Error>
+    where
+        Activity: ActivityHandler + Serialize + Debug,
+        Datatype: Clone,
+        ActorType: Actor,
+    {
+        build_tasks(activity, actor, inboxes, data).await
+    }
+
+    /// convert a sendactivitydata to a request, signing and sending it
+    pub async fn sign_and_send<Datatype: Clone>(&self, data: &Data<Datatype>) -> Result<(), Error> {
+        self.sign_and_send_internal(&data.config.client, data.config.request_timeout)
+            .await
+    }
+
+    pub(crate) async fn sign_and_send_internal(
+        &self,
+        client: &ClientWithMiddleware,
+        timeout: Duration,
+    ) -> Result<(), Error> {
+        debug!("Sending {} to {}", self.activity_id, self.inbox,);
+        let request_builder = client
+            .post(self.inbox.to_string())
+            .timeout(timeout)
+            .headers(generate_request_headers(&self.inbox));
+        let request = sign_request(
+            request_builder,
+            &self.actor_id,
+            self.activity.clone(),
+            self.private_key.clone(),
+            self.http_signature_compat,
+        )
+        .await?;
+        let response = client.execute(request).await?;
+        self.handle_response(response).await
+    }
+
+    /// Based on the HTTP status code determines if an activity was delivered successfully. In that case
+    /// Ok is returned. Otherwise it returns Err and the activity send should be retried later.
+    ///
+    /// Equivalent code in mastodon: https://github.com/mastodon/mastodon/blob/v4.2.8/app/helpers/jsonld_helper.rb#L215-L217
+    async fn handle_response(&self, response: Response) -> Result<(), Error> {
+        match response.status() {
+            status if status.is_success() => {
+                debug!("Activity {self} delivered successfully");
+                Ok(())
+            }
+            status
+                if status.is_client_error()
+                    && status != StatusCode::REQUEST_TIMEOUT
+                    && status != StatusCode::TOO_MANY_REQUESTS =>
+            {
+                let text = response.text_limited().await?;
+                debug!("Activity {self} was rejected, aborting: {text}");
+                Ok(())
+            }
+            status => {
+                let text = response.text_limited().await?;
+
+                Err(Error::Other(format!(
+                    "Activity {self} failure with status {status}: {text}",
+                )))
+            }
+        }
+    }
+}
+
+pub(crate) async fn build_tasks<'a, Activity, Datatype, ActorType>(
+    activity: &'a Activity,
+    actor: &ActorType,
+    inboxes: Vec<Url>,
+    data: &Data<Datatype>,
+) -> Result<Vec<SendActivityTask>, Error>
+where
+    Activity: ActivityHandler + Serialize + Debug,
+    Datatype: Clone,
+    ActorType: Actor,
+{
+    let config = &data.config;
+    let actor_id = activity.actor();
+    let activity_id = activity.id();
+    let activity_serialized: Bytes = serde_json::to_vec(activity)
+        .map_err(|e| Error::SerializeOutgoingActivity(e, format!("{:?}", activity)))?
+        .into();
+    let private_key = get_pkey_cached(data, actor).await?;
+
+    Ok(futures::stream::iter(
+        inboxes
+            .into_iter()
+            .unique()
+            .filter(|i| !config.is_local_url(i)),
+    )
+    .filter_map(|inbox| async {
+        if let Err(err) = config.verify_url_valid(&inbox).await {
+            debug!("inbox url invalid, skipping: {inbox}: {err}");
+            return None;
+        };
+        Some(SendActivityTask {
+            actor_id: actor_id.clone(),
+            activity_id: activity_id.clone(),
+            inbox,
+            activity: activity_serialized.clone(),
+            private_key: private_key.clone(),
+            http_signature_compat: config.http_signature_compat,
+        })
+    })
+    .collect()
+    .await)
+}
+
+pub(crate) async fn get_pkey_cached<ActorType>(
+    data: &Data<impl Clone>,
+    actor: &ActorType,
+) -> Result<PKey<Private>, Error>
+where
+    ActorType: Actor,
+{
+    let actor_id = actor.id();
+    // PKey is internally like an Arc<>, so cloning is ok
+    data.config
+        .actor_pkey_cache
+        .try_get_with_by_ref(&actor_id, async {
+            let private_key_pem = actor.private_key_pem().ok_or_else(|| {
+                Error::Other(format!(
+                    "Actor {actor_id} does not contain a private key for signing"
+                ))
+            })?;
+
+            // This is a mostly expensive blocking call, we don't want to tie up other tasks while this is happening
+            let pkey = tokio::task::spawn_blocking(move || {
+                PKey::private_key_from_pem(private_key_pem.as_bytes()).map_err(|err| {
+                    Error::Other(format!("Could not create private key from PEM data:{err}"))
+                })
+            })
+            .await
+            .map_err(|err| Error::Other(format!("Error joining: {err}")))??;
+            std::result::Result::<PKey<Private>, Error>::Ok(pkey)
+        })
+        .await
+        .map_err(|e| Error::Other(format!("cloned error: {e}")))
+}
+
+pub(crate) fn generate_request_headers(inbox_url: &Url) -> HeaderMap {
+    let mut host = inbox_url.domain().expect("read inbox domain").to_string();
+    if let Some(port) = inbox_url.port() {
+        host = format!("{}:{}", host, port);
+    }
+
+    let mut headers = HeaderMap::new();
+    headers.insert(
+        HeaderName::from_static("content-type"),
+        HeaderValue::from_static(FEDERATION_CONTENT_TYPE),
+    );
+    headers.insert(
+        HeaderName::from_static("host"),
+        HeaderValue::from_str(&host).expect("Hostname is valid"),
+    );
+    headers.insert(
+        "date",
+        HeaderValue::from_str(&fmt_http_date(SystemTime::now())).expect("Date is valid"),
+    );
+    headers
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use crate::{config::FederationConfig, http_signatures::generate_actor_keypair};
+    use std::{
+        sync::{atomic::AtomicUsize, Arc},
+        time::Instant,
+    };
+    use tracing::info;
+
+    // This will periodically send back internal errors to test the retry
+    async fn dodgy_handler(headers: HeaderMap, body: Bytes) -> Result<(), StatusCode> {
+        debug!("Headers:{:?}", headers);
+        debug!("Body len:{}", body.len());
+        Ok(())
+    }
+
+    async fn test_server() {
+        use axum::{routing::post, Router};
+
+        // We should break every now and then ;)
+        let state = Arc::new(AtomicUsize::new(0));
+
+        let app = Router::new()
+            .route("/", post(dodgy_handler))
+            .with_state(state);
+
+        axum::Server::bind(&"0.0.0.0:8001".parse().unwrap())
+            .serve(app.into_make_service())
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    // Sends 100 messages
+    async fn test_activity_sending() -> anyhow::Result<()> {
+        let num_messages: usize = 100;
+
+        tokio::spawn(test_server());
+
+        /*
+        // uncomment for debug logs & stats
+        use tracing::log::LevelFilter;
+
+        env_logger::builder()
+            .filter_level(LevelFilter::Warn)
+            .filter_module("activitypub_federation", LevelFilter::Info)
+            .format_timestamp(None)
+            .init();
+
+        */
+        let keypair = generate_actor_keypair().unwrap();
+
+        let message = SendActivityTask {
+            actor_id: "http://localhost:8001".parse().unwrap(),
+            activity_id: "http://localhost:8001/activity".parse().unwrap(),
+            activity: "{}".into(),
+            inbox: "http://localhost:8001".parse().unwrap(),
+            private_key: keypair.private_key().unwrap(),
+            http_signature_compat: true,
+        };
+        let data = FederationConfig::builder()
+            .app_data(())
+            .domain("localhost")
+            .build()
+            .await?
+            .to_request_data();
+
+        let start = Instant::now();
+
+        for _ in 0..num_messages {
+            message.clone().sign_and_send(&data).await?;
+        }
+
+        info!("Queue Sent: {:?}", start.elapsed());
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_handle_response() {
+        let keypair = generate_actor_keypair().unwrap();
+        let message = SendActivityTask {
+            actor_id: "http://localhost:8001".parse().unwrap(),
+            activity_id: "http://localhost:8001/activity".parse().unwrap(),
+            activity: "{}".into(),
+            inbox: "http://localhost:8001".parse().unwrap(),
+            private_key: keypair.private_key().unwrap(),
+            http_signature_compat: true,
+        };
+
+        let res = |status| {
+            http::Response::builder()
+                .status(status)
+                .body(vec![])
+                .unwrap()
+                .into()
+        };
+
+        assert!(message.handle_response(res(StatusCode::OK)).await.is_ok());
+        assert!(message
+            .handle_response(res(StatusCode::BAD_REQUEST))
+            .await
+            .is_ok());
+
+        assert!(message
+            .handle_response(res(StatusCode::MOVED_PERMANENTLY))
+            .await
+            .is_err());
+        assert!(message
+            .handle_response(res(StatusCode::REQUEST_TIMEOUT))
+            .await
+            .is_err());
+        assert!(message
+            .handle_response(res(StatusCode::TOO_MANY_REQUESTS))
+            .await
+            .is_err());
+        assert!(message
+            .handle_response(res(StatusCode::INTERNAL_SERVER_ERROR))
+            .await
+            .is_err());
+    }
+}
--- a/src/actix_web/inbox.rs
+++ b/src/actix_web/inbox.rs
@ -3,8 +3,8 @@
 use crate::{
    config::Data,
    error::Error,
-    fetch::object_id::ObjectId,
    http_signatures::{verify_body_hash, verify_signature},
+    parse_received_activity,
    traits::{ActivityHandler, Actor, Object},
 };
 use actix_web::{web::Bytes, HttpRequest, HttpResponse};
@ -23,20 +23,13 @@ where
    Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
    ActorT: Object<DataType = Datatype> + Actor + Send + 'static,
    for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
-    <Activity as ActivityHandler>::Error: From<anyhow::Error>
-        + From<Error>
-        + From<<ActorT as Object>::Error>
-        + From<serde_json::Error>,
-    <ActorT as Object>::Error: From<Error> + From<anyhow::Error>,
+    <Activity as ActivityHandler>::Error: From<Error> + From<<ActorT as Object>::Error>,
+    <ActorT as Object>::Error: From<Error>,
    Datatype: Clone,
 {
    verify_body_hash(request.headers().get("Digest"), &body)?;

-    let activity: Activity = serde_json::from_slice(&body)?;
-    data.config.verify_url_and_domain(&activity).await?;
-    let actor = ObjectId::<ActorT>::from(activity.actor().clone())
-        .dereference(data)
-        .await?;
+    let (activity, actor) = parse_received_activity::<Activity, ActorT, _>(&body, data).await?;

    verify_signature(
        request.headers(),
@ -52,17 +45,20 @@ where
 }

 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod test {
    use super::*;
    use crate::{
-        activity_queue::generate_request_headers,
+        activity_sending::generate_request_headers,
        config::FederationConfig,
+        fetch::object_id::ObjectId,
        http_signatures::sign_request,
        traits::tests::{DbConnection, DbUser, Follow, DB_USER_KEYPAIR},
    };
    use actix_web::test::TestRequest;
    use reqwest::Client;
    use reqwest_middleware::ClientWithMiddleware;
+    use serde_json::json;
    use url::Url;

    #[tokio::test]
@ -89,8 +85,7 @@ mod test {
        .err()
        .unwrap();

-        let e = err.root_cause().downcast_ref::<Error>().unwrap();
-        assert_eq!(e, &Error::ActivityBodyDigestInvalid)
+        assert_eq!(&err, &Error::ActivityBodyDigestInvalid)
    }

    #[tokio::test]
@ -106,26 +101,52 @@ mod test {
        .err()
        .unwrap();

-        let e = err.root_cause().downcast_ref::<Error>().unwrap();
-        assert_eq!(e, &Error::ActivitySignatureInvalid)
+        assert_eq!(&err, &Error::ActivitySignatureInvalid)
    }

-    async fn setup_receive_test() -> (Bytes, TestRequest, FederationConfig<DbConnection>) {
+    #[tokio::test]
+    async fn test_receive_unparseable_activity() {
+        let (_, _, config) = setup_receive_test().await;
+
+        let actor = Url::parse("http://ds9.lemmy.ml/u/lemmy_alpha").unwrap();
+        let id = "http://localhost:123/1";
+        let activity = json!({
+          "actor": actor.as_str(),
+          "to": ["https://www.w3.org/ns/activitystreams#Public"],
+          "object": "http://ds9.lemmy.ml/post/1",
+          "cc": ["http://enterprise.lemmy.ml/c/main"],
+          "type": "Delete",
+          "id": id
+        }
+        );
+        let body: Bytes = serde_json::to_vec(&activity).unwrap().into();
+        let incoming_request = construct_request(&body, &actor).await;
+
+        // intentionally cause a parse error by using wrong type for deser
+        let res = receive_activity::<Follow, DbUser, DbConnection>(
+            incoming_request.to_http_request(),
+            body,
+            &config.to_request_data(),
+        )
+        .await;
+
+        match res {
+            Err(Error::ParseReceivedActivity(_, url)) => {
+                assert_eq!(id, url.expect("has url").as_str());
+            }
+            _ => unreachable!(),
+        }
+    }
+
+    async fn construct_request(body: &Bytes, actor: &Url) -> TestRequest {
        let inbox = "https://example.com/inbox";
        let headers = generate_request_headers(&Url::parse(inbox).unwrap());
        let request_builder = ClientWithMiddleware::from(Client::default())
            .post(inbox)
            .headers(headers);
-        let activity = Follow {
-            actor: ObjectId::parse("http://localhost:123").unwrap(),
-            object: ObjectId::parse("http://localhost:124").unwrap(),
-            kind: Default::default(),
-            id: "http://localhost:123/1".try_into().unwrap(),
-        };
-        let body: Bytes = serde_json::to_vec(&activity).unwrap().into();
        let outgoing_request = sign_request(
            request_builder,
-            &activity.actor.into_inner(),
+            actor,
            body.clone(),
            DB_USER_KEYPAIR.private_key().unwrap(),
            false,
@ -136,6 +157,18 @@ mod test {
        for h in outgoing_request.headers() {
            incoming_request = incoming_request.append_header(h);
        }
+        incoming_request
+    }
+
+    async fn setup_receive_test() -> (Bytes, TestRequest, FederationConfig<DbConnection>) {
+        let activity = Follow {
+            actor: ObjectId::parse("http://localhost:123").unwrap(),
+            object: ObjectId::parse("http://localhost:124").unwrap(),
+            kind: Default::default(),
+            id: "http://localhost:123/1".try_into().unwrap(),
+        };
+        let body: Bytes = serde_json::to_vec(&activity).unwrap().into();
+        let incoming_request = construct_request(&body, activity.actor.inner()).await;

        let config = FederationConfig::builder()
            .domain("localhost:8002")
--- a/src/actix_web/mod.rs
+++ b/src/actix_web/mod.rs
@ -22,7 +22,7 @@ pub async fn signing_actor<A>(
 ) -> Result<A, <A as Object>::Error>
 where
    A: Object + Actor,
-    <A as Object>::Error: From<Error> + From<anyhow::Error>,
+    <A as Object>::Error: From<Error>,
    for<'de2> <A as Object>::Kind: Deserialize<'de2>,
 {
    verify_body_hash(request.headers().get("Digest"), &body.unwrap_or_default())?;
--- a/src/axum/inbox.rs
+++ b/src/axum/inbox.rs
@ -5,8 +5,8 @@
 use crate::{
    config::Data,
    error::Error,
-    fetch::object_id::ObjectId,
-    http_signatures::{verify_body_hash, verify_signature},
+    http_signatures::verify_signature,
+    parse_received_activity,
    traits::{ActivityHandler, Actor, Object},
 };
 use axum::{
@ -29,20 +29,12 @@ where
    Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
    ActorT: Object<DataType = Datatype> + Actor + Send + 'static,
    for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
-    <Activity as ActivityHandler>::Error: From<anyhow::Error>
-        + From<Error>
-        + From<<ActorT as Object>::Error>
-        + From<serde_json::Error>,
-    <ActorT as Object>::Error: From<Error> + From<anyhow::Error>,
+    <Activity as ActivityHandler>::Error: From<Error> + From<<ActorT as Object>::Error>,
+    <ActorT as Object>::Error: From<Error>,
    Datatype: Clone,
 {
-    verify_body_hash(activity_data.headers.get("Digest"), &activity_data.body)?;
-
-    let activity: Activity = serde_json::from_slice(&activity_data.body)?;
-    data.config.verify_url_and_domain(&activity).await?;
-    let actor = ObjectId::<ActorT>::from(activity.actor().clone())
-        .dereference(data)
-        .await?;
+    let (activity, actor) =
+        parse_received_activity::<Activity, ActorT, _>(&activity_data.body, data).await?;

    verify_signature(
        &activity_data.headers,
--- a/src/axum/json.rs
+++ b/src/axum/json.rs
@ -9,7 +9,7 @@
 //! # use activitypub_federation::traits::Object;
 //! # use activitypub_federation::traits::tests::{DbConnection, DbUser, Person};
 //! async fn http_get_user(Path(name): Path<String>, data: Data<DbConnection>) -> Result<FederationJson<WithContext<Person>>, Error> {
-//!     let user: DbUser = data.read_local_user(name).await?;
+//!     let user: DbUser = data.read_local_user(&name).await?;
 //!     let person = user.into_json(&data).await?;
 //!
 //!     Ok(FederationJson(WithContext::new_default(person)))
--- a/src/config.rs
+++ b/src/config.rs
@ -9,7 +9,6 @@
 //!     .domain("example.com")
 //!     .app_data(())
 //!     .http_fetch_limit(50)
-//!     .worker_count(16)
 //!     .build().await?;
 //! # Ok::<(), anyhow::Error>(())
 //! # }).unwrap()
@ -24,6 +23,7 @@ use crate::{
 use async_trait::async_trait;
 use derive_builder::Builder;
 use dyn_clone::{clone_trait_object, DynClone};
+use moka::future::Cache;
 use openssl::pkey::{PKey, Private};
 use reqwest_middleware::ClientWithMiddleware;
 use serde::de::DeserializeOwned;
@ -55,19 +55,14 @@ pub struct FederationConfig<T: Clone> {
    /// HTTP client used for all outgoing requests. Middleware can be used to add functionality
    /// like log tracing or retry of failed requests.
    pub(crate) client: ClientWithMiddleware,
-    /// Number of tasks that can be in-flight concurrently.
-    /// Tasks are retried once after a minute, then put into the retry queue
-    #[builder(default = "1024")]
-    pub(crate) worker_count: usize,
-    /// Number of concurrent tasks that are being retried in-flight concurrently.
-    /// Tasks are retried after an hour, then again in 60 hours.
-    #[builder(default = "128")]
-    pub(crate) retry_count: usize,
    /// Run library in debug mode. This allows usage of http and localhost urls. It also sends
    /// outgoing activities synchronously, not in background thread. This helps to make tests
    /// more consistent. Do not use for production.
    #[builder(default = "false")]
    pub(crate) debug: bool,
+    /// Allow HTTP urls even in production mode
+    #[builder(default = "self.debug.unwrap_or(false)")]
+    pub(crate) allow_http_urls: bool,
    /// Timeout for all HTTP requests. HTTP signatures are valid for 10s, so it makes sense to
    /// use the same as timeout when sending
    #[builder(default = "Duration::from_secs(10)")]
@ -86,10 +81,25 @@ pub struct FederationConfig<T: Clone> {
    /// <https://docs.joinmastodon.org/spec/activitypub/#secure-mode>
    #[builder(default = "None", setter(custom))]
    pub(crate) signed_fetch_actor: Option<Arc<(Url, PKey<Private>)>>,
+    #[builder(
+        default = "Cache::builder().max_capacity(10000).build()",
+        setter(custom)
+    )]
+    pub(crate) actor_pkey_cache: Cache<Url, PKey<Private>>,
    /// Queue for sending outgoing activities. Only optional to make builder work, its always
    /// present once constructed.
    #[builder(setter(skip))]
    pub(crate) activity_queue: Option<Arc<ActivityQueue>>,
+    /// When sending with activity queue: Number of tasks that can be in-flight concurrently.
+    /// Tasks are retried once after a minute, then put into the retry queue.
+    /// Setting this count to `0` means that there is no limit to concurrency
+    #[builder(default = "0")]
+    pub(crate) queue_worker_count: usize,
+    /// When sending with activity queue: Number of concurrent tasks that are being retried
+    /// in-flight concurrently. Tasks are retried after an hour, then again in 60 hours.
+    /// Setting this count to `0` means that there is no limit to concurrency
+    #[builder(default = "0")]
+    pub(crate) queue_retry_count: usize,
 }

 impl<T: Clone> FederationConfig<T> {
@ -132,7 +142,7 @@ impl<T: Clone> FederationConfig<T> {
        match url.scheme() {
            "https" => {}
            "http" => {
-                if !self.debug {
+                if !self.allow_http_urls {
                    return Err(Error::UrlVerificationError(
                        "Http urls are only allowed in debug mode",
                    ));
@ -156,10 +166,7 @@ impl<T: Clone> FederationConfig<T> {
            ));
        }

-        self.url_verifier
-            .verify(url)
-            .await
-            .map_err(Error::UrlVerificationError)?;
+        self.url_verifier.verify(url).await?;

        Ok(())
    }
@ -167,11 +174,17 @@ impl<T: Clone> FederationConfig<T> {
    /// Returns true if the url refers to this instance. Handles hostnames like `localhost:8540` for
    /// local debugging.
    pub(crate) fn is_local_url(&self, url: &Url) -> bool {
-        let mut domain = url.host_str().expect("id has domain").to_string();
-        if let Some(port) = url.port() {
-            domain = format!("{}:{}", domain, port);
+        match url.host_str() {
+            Some(domain) => {
+                let domain = if let Some(port) = url.port() {
+                    format!("{}:{}", domain, port)
+                } else {
+                    domain.to_string()
+                };
+                domain == self.domain
+            }
+            None => false,
        }
-        domain == self.domain
    }

    /// Returns the local domain
@ -193,6 +206,12 @@ impl<T: Clone> FederationConfigBuilder<T> {
        self
    }

+    /// sets the number of parsed actor private keys to keep in memory
+    pub fn actor_pkey_cache(&mut self, cache_size: u64) -> &mut Self {
+        self.actor_pkey_cache = Some(Cache::builder().max_capacity(cache_size).build());
+        self
+    }
+
    /// Constructs a new config instance with the values supplied to builder.
    ///
    /// Values which are not explicitly specified use the defaults. Also initializes the
@ -202,8 +221,8 @@ impl<T: Clone> FederationConfigBuilder<T> {
        let mut config = self.partial_build()?;
        let queue = create_activity_queue(
            config.client.clone(),
-            config.worker_count,
-            config.retry_count,
+            config.queue_worker_count,
+            config.queue_retry_count,
            config.request_timeout,
        );
        config.activity_queue = Some(Arc::new(queue));
@ -230,6 +249,7 @@ impl<T: Clone> Deref for FederationConfig<T> {
 /// # use async_trait::async_trait;
 /// # use url::Url;
 /// # use activitypub_federation::config::UrlVerifier;
+/// # use activitypub_federation::error::Error;
 /// # #[derive(Clone)]
 /// # struct DatabaseConnection();
 /// # async fn get_blocklist(_: &DatabaseConnection) -> Vec<String> {
@ -242,11 +262,11 @@ impl<T: Clone> Deref for FederationConfig<T> {
 ///
 /// #[async_trait]
 /// impl UrlVerifier for Verifier {
-///     async fn verify(&self, url: &Url) -> Result<(), &'static str> {
+///     async fn verify(&self, url: &Url) -> Result<(), Error> {
 ///         let blocklist = get_blocklist(&self.db_connection).await;
 ///         let domain = url.domain().unwrap().to_string();
 ///         if blocklist.contains(&domain) {
-///             Err("Domain is blocked")
+///             Err(Error::Other("Domain is blocked".into()))
 ///         } else {
 ///             Ok(())
 ///         }
@ -256,7 +276,7 @@ impl<T: Clone> Deref for FederationConfig<T> {
 #[async_trait]
 pub trait UrlVerifier: DynClone + Send {
    /// Should return Ok iff the given url is valid for processing.
-    async fn verify(&self, url: &Url) -> Result<(), &'static str>;
+    async fn verify(&self, url: &Url) -> Result<(), Error>;
 }

 /// Default URL verifier which does nothing.
@ -265,7 +285,7 @@ struct DefaultUrlVerifier();

 #[async_trait]
 impl UrlVerifier for DefaultUrlVerifier {
-    async fn verify(&self, _url: &Url) -> Result<(), &'static str> {
+    async fn verify(&self, _url: &Url) -> Result<(), Error> {
        Ok(())
    }
 }
@ -327,3 +347,34 @@ impl<T: Clone> FederationMiddleware<T> {
        FederationMiddleware(config)
    }
 }
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod test {
+    use super::*;
+
+    async fn config() -> FederationConfig<i32> {
+        FederationConfig::builder()
+            .domain("example.com")
+            .app_data(1)
+            .build()
+            .await
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_url_is_local() -> Result<(), Error> {
+        let config = config().await;
+        assert!(config.is_local_url(&Url::parse("http://example.com")?));
+        assert!(!config.is_local_url(&Url::parse("http://other.com")?));
+        // ensure that missing domain doesnt cause crash
+        assert!(!config.is_local_url(&Url::parse("http://127.0.0.1")?));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_get_domain() {
+        let config = config().await;
+        assert_eq!("example.com", config.domain());
+    }
+}
--- a/src/error.rs
+++ b/src/error.rs
@ -1,37 +1,88 @@
 //! Error messages returned by this library

-use displaydoc::Display;
+use crate::fetch::webfinger::WebFingerError;
+use http_signature_normalization_reqwest::SignError;
+use openssl::error::ErrorStack;
+use std::string::FromUtf8Error;
+use tokio::task::JoinError;
+use url::Url;

 /// Error messages returned by this library
-#[derive(thiserror::Error, Debug, Display)]
+#[derive(thiserror::Error, Debug)]
 pub enum Error {
    /// Object was not found in local database
+    #[error("Object was not found in local database")]
    NotFound,
    /// Request limit was reached during fetch
+    #[error("Request limit was reached during fetch")]
    RequestLimit,
    /// Response body limit was reached during fetch
+    #[error("Response body limit was reached during fetch")]
    ResponseBodyLimit,
    /// Object to be fetched was deleted
-    ObjectDeleted,
-    /// {0}
+    #[error("Fetched remote object {0} which was deleted")]
+    ObjectDeleted(Url),
+    /// url verification error
+    #[error("URL failed verification: {0}")]
    UrlVerificationError(&'static str),
    /// Incoming activity has invalid digest for body
+    #[error("Incoming activity has invalid digest for body")]
    ActivityBodyDigestInvalid,
    /// Incoming activity has invalid signature
+    #[error("Incoming activity has invalid signature")]
    ActivitySignatureInvalid,
    /// Failed to resolve actor via webfinger
-    WebfingerResolveFailed,
-    /// Other errors which are not explicitly handled
+    #[error("Failed to resolve actor via webfinger")]
+    WebfingerResolveFailed(#[from] WebFingerError),
+    /// Failed to serialize outgoing activity
+    #[error("Failed to serialize outgoing activity {1}: {0}")]
+    SerializeOutgoingActivity(serde_json::Error, String),
+    /// Failed to parse an object fetched from url
+    #[error("Failed to parse object {1} with content {2}: {0}")]
+    ParseFetchedObject(serde_json::Error, Url, String),
+    /// Failed to parse an activity received from another instance
+    #[error("Failed to parse incoming activity {}: {0}", match .1 {
+        Some(t) => format!("with id {t}"),
+        None => String::new(),
+    })]
+    ParseReceivedActivity(serde_json::Error, Option<Url>),
+    /// Reqwest Middleware Error
    #[error(transparent)]
-    Other(#[from] anyhow::Error),
+    ReqwestMiddleware(#[from] reqwest_middleware::Error),
+    /// Reqwest Error
+    #[error(transparent)]
+    Reqwest(#[from] reqwest::Error),
+    /// UTF-8 error
+    #[error(transparent)]
+    Utf8(#[from] FromUtf8Error),
+    /// Url Parse
+    #[error(transparent)]
+    UrlParse(#[from] url::ParseError),
+    /// Signing errors
+    #[error(transparent)]
+    SignError(#[from] SignError),
+    /// Failed to queue activity for sending
+    #[error("Failed to queue activity {0} for sending")]
+    ActivityQueueError(Url),
+    /// Stop activity queue
+    #[error(transparent)]
+    StopActivityQueue(#[from] JoinError),
+    /// Attempted to fetch object which doesn't have valid ActivityPub Content-Type
+    #[error(
+        "Attempted to fetch object from {0} which doesn't have valid ActivityPub Content-Type"
+    )]
+    FetchInvalidContentType(Url),
+    /// Attempted to fetch object but the response's id field doesn't match
+    #[error("Attempted to fetch object from {0} but the response's id field doesn't match")]
+    FetchWrongId(Url),
+    /// Other generic errors
+    #[error("{0}")]
+    Other(String),
 }

-impl Error {
-    pub(crate) fn other<T>(error: T) -> Self
-    where
-        T: Into<anyhow::Error>,
-    {
-        Error::Other(error.into())
+impl From<ErrorStack> for Error {
+    fn from(value: ErrorStack) -> Self {
+        Error::Other(value.to_string())
    }
 }

--- a/src/fetch/collection_id.rs
+++ b/src/fetch/collection_id.rs
@ -20,12 +20,8 @@ where
    for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
 {
    /// Construct a new CollectionId instance
-    pub fn parse<T>(url: T) -> Result<Self, url::ParseError>
-    where
-        T: TryInto<Url>,
-        url::ParseError: From<<T as TryInto<Url>>::Error>,
-    {
-        Ok(Self(Box::new(url.try_into()?), PhantomData::<Kind>))
+    pub fn parse(url: &str) -> Result<Self, url::ParseError> {
+        Ok(Self(Box::new(Url::parse(url)?), PhantomData::<Kind>))
    }

    /// Fetches collection over HTTP
@ -40,9 +36,10 @@ where
    where
        <Kind as Collection>::Error: From<Error>,
    {
-        let json = fetch_object_http(&self.0, data).await?;
-        Kind::verify(&json, &self.0, data).await?;
-        Kind::from_json(json, owner, data).await
+        let res = fetch_object_http(&self.0, data).await?;
+        let redirect_url = &res.url;
+        Kind::verify(&res.object, redirect_url, data).await?;
+        Kind::from_json(res.object, owner, data).await
    }
 }

@ -95,3 +92,102 @@ where
        CollectionId(Box::new(url), PhantomData::<Kind>)
    }
 }
+
+impl<Kind> PartialEq for CollectionId<Kind>
+where
+    Kind: Collection,
+    for<'de2> <Kind as Collection>::Kind: serde::Deserialize<'de2>,
+{
+    fn eq(&self, other: &Self) -> bool {
+        self.0.eq(&other.0) && self.1 == other.1
+    }
+}
+
+#[cfg(feature = "diesel")]
+const _IMPL_DIESEL_NEW_TYPE_FOR_COLLECTION_ID: () = {
+    use diesel::{
+        backend::Backend,
+        deserialize::{FromSql, FromStaticSqlRow},
+        expression::AsExpression,
+        internal::derives::as_expression::Bound,
+        pg::Pg,
+        query_builder::QueryId,
+        serialize,
+        serialize::{Output, ToSql},
+        sql_types::{HasSqlType, SingleValue, Text},
+        Expression,
+        Queryable,
+    };
+
+    // TODO: this impl only works for Postgres db because of to_string() call which requires reborrow
+    impl<Kind, ST> ToSql<ST, Pg> for CollectionId<Kind>
+    where
+        Kind: Collection,
+        for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
+        String: ToSql<ST, Pg>,
+    {
+        fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result {
+            let v = self.0.to_string();
+            <String as ToSql<Text, Pg>>::to_sql(&v, &mut out.reborrow())
+        }
+    }
+    impl<'expr, Kind, ST> AsExpression<ST> for &'expr CollectionId<Kind>
+    where
+        Kind: Collection,
+        for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
+        Bound<ST, String>: Expression<SqlType = ST>,
+        ST: SingleValue,
+    {
+        type Expression = Bound<ST, &'expr str>;
+        fn as_expression(self) -> Self::Expression {
+            Bound::new(self.0.as_str())
+        }
+    }
+    impl<Kind, ST> AsExpression<ST> for CollectionId<Kind>
+    where
+        Kind: Collection,
+        for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
+        Bound<ST, String>: Expression<SqlType = ST>,
+        ST: SingleValue,
+    {
+        type Expression = Bound<ST, String>;
+        fn as_expression(self) -> Self::Expression {
+            Bound::new(self.0.to_string())
+        }
+    }
+    impl<Kind, ST, DB> FromSql<ST, DB> for CollectionId<Kind>
+    where
+        Kind: Collection + Send + 'static,
+        for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
+        String: FromSql<ST, DB>,
+        DB: Backend,
+        DB: HasSqlType<ST>,
+    {
+        fn from_sql(
+            raw: DB::RawValue<'_>,
+        ) -> Result<Self, Box<dyn ::std::error::Error + Send + Sync>> {
+            let string: String = FromSql::<ST, DB>::from_sql(raw)?;
+            Ok(CollectionId::parse(&string)?)
+        }
+    }
+    impl<Kind, ST, DB> Queryable<ST, DB> for CollectionId<Kind>
+    where
+        Kind: Collection + Send + 'static,
+        for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
+        String: FromStaticSqlRow<ST, DB>,
+        DB: Backend,
+        DB: HasSqlType<ST>,
+    {
+        type Row = String;
+        fn build(row: Self::Row) -> diesel::deserialize::Result<Self> {
+            Ok(CollectionId::parse(&row)?)
+        }
+    }
+    impl<Kind> QueryId for CollectionId<Kind>
+    where
+        Kind: Collection + 'static,
+        for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
+    {
+        type QueryId = Self;
+    }
+};
--- a/src/fetch/mod.rs
+++ b/src/fetch/mod.rs
@ -4,13 +4,14 @@

 use crate::{
    config::Data,
-    error::Error,
+    error::{Error, Error::ParseFetchedObject},
+    extract_id,
    http_signatures::sign_request,
    reqwest_shim::ResponseExt,
    FEDERATION_CONTENT_TYPE,
 };
 use bytes::Bytes;
-use http::StatusCode;
+use http::{HeaderValue, StatusCode};
 use serde::de::DeserializeOwned;
 use std::sync::atomic::Ordering;
 use tracing::info;
@ -23,6 +24,16 @@ pub mod object_id;
 /// Resolves identifiers of the form `name@example.com`
 pub mod webfinger;

+/// Response from fetching a remote object
+pub struct FetchObjectResponse<Kind> {
+    /// The resolved object
+    pub object: Kind,
+    /// Contains the final URL (different from request URL in case of redirect)
+    pub url: Url,
+    content_type: Option<HeaderValue>,
+    object_id: Option<Url>,
+}
+
 /// Fetch a remote object over HTTP and convert to `Kind`.
 ///
 /// [crate::fetch::object_id::ObjectId::dereference] wraps this function to add caching and
@ -33,17 +44,61 @@ pub mod webfinger;
 /// If the value exceeds [FederationSettings.http_fetch_limit], the request is aborted with
 /// [Error::RequestLimit]. This prevents denial of service attacks where an attack triggers
 /// infinite, recursive fetching of data.
+///
+/// The `Accept` header will be set to the content of [`FEDERATION_CONTENT_TYPE`]. When parsing the
+/// response it ensures that it has a valid `Content-Type` header as defined by ActivityPub, to
+/// prevent security vulnerabilities like [this one](https://github.com/mastodon/mastodon/security/advisories/GHSA-jhrq-qvrm-qr36).
+/// Additionally it checks that the `id` field is identical to the fetch URL (after redirects).
 pub async fn fetch_object_http<T: Clone, Kind: DeserializeOwned>(
    url: &Url,
    data: &Data<T>,
-) -> Result<Kind, Error> {
+) -> Result<FetchObjectResponse<Kind>, Error> {
+    static FETCH_CONTENT_TYPE: HeaderValue = HeaderValue::from_static(FEDERATION_CONTENT_TYPE);
+    const VALID_RESPONSE_CONTENT_TYPES: [&str; 3] = [
+        FEDERATION_CONTENT_TYPE, // lemmy
+        r#"application/ld+json; profile="https://www.w3.org/ns/activitystreams""#, // activitypub standard
+        r#"application/activity+json; charset=utf-8"#,                             // mastodon
+    ];
+    let res = fetch_object_http_with_accept(url, data, &FETCH_CONTENT_TYPE).await?;
+
+    // Ensure correct content-type to prevent vulnerabilities, with case insensitive comparison.
+    let content_type = res
+        .content_type
+        .as_ref()
+        .and_then(|c| c.to_str().ok())
+        .ok_or(Error::FetchInvalidContentType(res.url.clone()))?;
+    if !VALID_RESPONSE_CONTENT_TYPES.contains(&content_type) {
+        return Err(Error::FetchInvalidContentType(res.url));
+    }
+
+    // Ensure id field matches final url after redirect
+    if res.object_id.as_ref() != Some(&res.url) {
+        return Err(Error::FetchWrongId(res.url));
+    }
+
+    // Dont allow fetching local object. Only check this after the request as a local url
+    // may redirect to a remote object.
+    if data.config.is_local_url(&res.url) {
+        return Err(Error::NotFound);
+    }
+
+    Ok(res)
+}
+
+/// Fetch a remote object over HTTP and convert to `Kind`. This function works exactly as
+/// [`fetch_object_http`] except that the `Accept` header is specified in `content_type`.
+async fn fetch_object_http_with_accept<T: Clone, Kind: DeserializeOwned>(
+    url: &Url,
+    data: &Data<T>,
+    content_type: &HeaderValue,
+) -> Result<FetchObjectResponse<Kind>, Error> {
    let config = &data.config;
-    // dont fetch local objects this way
-    debug_assert!(url.domain() != Some(&config.domain));
    config.verify_url_valid(url).await?;
    info!("Fetching remote object {}", url.to_string());

-    let counter = data.request_counter.fetch_add(1, Ordering::SeqCst);
+    let mut counter = data.request_counter.fetch_add(1, Ordering::SeqCst);
+    // fetch_add returns old value so we need to increment manually here
+    counter += 1;
    if counter > config.http_fetch_limit {
        return Err(Error::RequestLimit);
    }
@ -51,7 +106,7 @@ pub async fn fetch_object_http<T: Clone, Kind: DeserializeOwned>(
    let req = config
        .client
        .get(url.as_str())
-        .header("Accept", FEDERATION_CONTENT_TYPE)
+        .header("Accept", content_type)
        .timeout(config.request_timeout);

    let res = if let Some((actor_id, private_key_pem)) = config.signed_fetch_actor.as_deref() {
@ -63,14 +118,31 @@ pub async fn fetch_object_http<T: Clone, Kind: DeserializeOwned>(
            data.config.http_signature_compat,
        )
        .await?;
-        config.client.execute(req).await.map_err(Error::other)?
+        config.client.execute(req).await?
    } else {
-        req.send().await.map_err(Error::other)?
+        req.send().await?
    };

    if res.status() == StatusCode::GONE {
-        return Err(Error::ObjectDeleted);
+        return Err(Error::ObjectDeleted(url.clone()));
    }

-    res.json_limited().await
+    let url = res.url().clone();
+    let content_type = res.headers().get("Content-Type").cloned();
+    let text = res.bytes_limited().await?;
+    let object_id = extract_id(&text).ok();
+
+    match serde_json::from_slice(&text) {
+        Ok(object) => Ok(FetchObjectResponse {
+            object,
+            url,
+            content_type,
+            object_id,
+        }),
+        Err(e) => Err(ParseFetchedObject(
+            e,
+            url,
+            String::from_utf8(Vec::from(text))?,
+        )),
+    }
 }
--- a/src/fetch/object_id.rs
+++ b/src/fetch/object_id.rs
@ -1,6 +1,5 @@
 use crate::{config::Data, error::Error, fetch::fetch_object_http, traits::Object};
-use anyhow::anyhow;
-use chrono::{Duration as ChronoDuration, NaiveDateTime, Utc};
+use chrono::{DateTime, Duration as ChronoDuration, Utc};
 use serde::{Deserialize, Serialize};
 use std::{
    fmt::{Debug, Display, Formatter},
@ -11,7 +10,7 @@ use url::Url;

 impl<T> FromStr for ObjectId<T>
 where
-    T: Object + Send + 'static,
+    T: Object + Send + Debug + 'static,
    for<'de2> <T as Object>::Kind: Deserialize<'de2>,
 {
    type Err = url::ParseError;
@ -58,20 +57,16 @@ where
 pub struct ObjectId<Kind>(Box<Url>, PhantomData<Kind>)
 where
    Kind: Object,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>;
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>;

 impl<Kind> ObjectId<Kind>
 where
-    Kind: Object + Send + 'static,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
+    Kind: Object + Send + Debug + 'static,
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
 {
    /// Construct a new objectid instance
-    pub fn parse<T>(url: T) -> Result<Self, url::ParseError>
-    where
-        T: TryInto<Url>,
-        url::ParseError: From<<T as TryInto<Url>>::Error>,
-    {
-        Ok(ObjectId(Box::new(url.try_into()?), PhantomData::<Kind>))
+    pub fn parse(url: &str) -> Result<Self, url::ParseError> {
+        Ok(Self(Box::new(Url::parse(url)?), PhantomData::<Kind>))
    }

    /// Returns a reference to the wrapped URL value
@ -90,23 +85,16 @@ where
        data: &Data<<Kind as Object>::DataType>,
    ) -> Result<Kind, <Kind as Object>::Error>
    where
-        <Kind as Object>::Error: From<Error> + From<anyhow::Error>,
+        <Kind as Object>::Error: From<Error>,
    {
        let db_object = self.dereference_from_db(data).await?;

-        // if its a local object, only fetch it from the database and not over http
-        if data.config.is_local_url(&self.0) {
-            return match db_object {
-                None => Err(Error::NotFound.into()),
-                Some(o) => Ok(o),
-            };
-        }
-
        // object found in database
        if let Some(object) = db_object {
-            // object is old and should be refetched
            if let Some(last_refreshed_at) = object.last_refreshed_at() {
-                if should_refetch_object(last_refreshed_at) {
+                let is_local = data.config.is_local_url(&self.0);
+                if !is_local && should_refetch_object(last_refreshed_at) {
+                    // object is outdated and should be refetched
                    return self.dereference_from_http(data, Some(object)).await;
                }
            }
@ -118,6 +106,24 @@ where
        }
    }

+    /// If this is a remote object, fetch it from origin instance unconditionally to get the
+    /// latest version, regardless of refresh interval.
+    pub async fn dereference_forced(
+        &self,
+        data: &Data<<Kind as Object>::DataType>,
+    ) -> Result<Kind, <Kind as Object>::Error>
+    where
+        <Kind as Object>::Error: From<Error>,
+    {
+        if data.config.is_local_url(&self.0) {
+            self.dereference_from_db(data)
+                .await
+                .map(|o| o.ok_or(Error::NotFound.into()))?
+        } else {
+            self.dereference_from_http(data, None).await
+        }
+    }
+
    /// Fetch an object from the local db. Instead of falling back to http, this throws an error if
    /// the object is not found in the database.
    pub async fn dereference_local(
@ -146,21 +152,27 @@ where
        db_object: Option<Kind>,
    ) -> Result<Kind, <Kind as Object>::Error>
    where
-        <Kind as Object>::Error: From<Error> + From<anyhow::Error>,
+        <Kind as Object>::Error: From<Error>,
    {
        let res = fetch_object_http(&self.0, data).await;

-        if let Err(Error::ObjectDeleted) = &res {
+        if let Err(Error::ObjectDeleted(url)) = res {
            if let Some(db_object) = db_object {
                db_object.delete(data).await?;
            }
-            return Err(anyhow!("Fetched remote object {} which was deleted", self).into());
+            return Err(Error::ObjectDeleted(url).into());
        }

-        let res2 = res?;
+        let res = res?;
+        let redirect_url = &res.url;

-        Kind::verify(&res2, self.inner(), data).await?;
-        Kind::from_json(res2, data).await
+        Kind::verify(&res.object, redirect_url, data).await?;
+        Kind::from_json(res.object, data).await
+    }
+
+    /// Returns true if the object's domain matches the one defined in [[FederationConfig.domain]].
+    pub fn is_local(&self, data: &Data<<Kind as Object>::DataType>) -> bool {
+        data.config.is_local_url(&self.0)
    }
 }

@ -168,7 +180,7 @@ where
 impl<Kind> Clone for ObjectId<Kind>
 where
    Kind: Object,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
 {
    fn clone(&self) -> Self {
        ObjectId(self.0.clone(), self.1)
@ -181,21 +193,21 @@ static ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG: i64 = 20;
 /// Determines when a remote actor should be refetched from its instance. In release builds, this is
 /// `ACTOR_REFETCH_INTERVAL_SECONDS` after the last refetch, in debug builds
 /// `ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG`.
-fn should_refetch_object(last_refreshed: NaiveDateTime) -> bool {
+fn should_refetch_object(last_refreshed: DateTime<Utc>) -> bool {
    let update_interval = if cfg!(debug_assertions) {
        // avoid infinite loop when fetching community outbox
-        ChronoDuration::seconds(ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG)
+        ChronoDuration::try_seconds(ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG).expect("valid duration")
    } else {
-        ChronoDuration::seconds(ACTOR_REFETCH_INTERVAL_SECONDS)
+        ChronoDuration::try_seconds(ACTOR_REFETCH_INTERVAL_SECONDS).expect("valid duration")
    };
-    let refresh_limit = Utc::now().naive_utc() - update_interval;
+    let refresh_limit = Utc::now() - update_interval;
    last_refreshed.lt(&refresh_limit)
 }

 impl<Kind> Display for ObjectId<Kind>
 where
    Kind: Object,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
 {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0.as_str())
@ -205,7 +217,7 @@ where
 impl<Kind> Debug for ObjectId<Kind>
 where
    Kind: Object,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
 {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0.as_str())
@ -215,7 +227,7 @@ where
 impl<Kind> From<ObjectId<Kind>> for Url
 where
    Kind: Object,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
 {
    fn from(id: ObjectId<Kind>) -> Self {
        *id.0
@ -225,7 +237,7 @@ where
 impl<Kind> From<Url> for ObjectId<Kind>
 where
    Kind: Object + Send + 'static,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
 {
    fn from(url: Url) -> Self {
        ObjectId(Box::new(url), PhantomData::<Kind>)
@ -235,17 +247,107 @@ where
 impl<Kind> PartialEq for ObjectId<Kind>
 where
    Kind: Object,
-    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
+    for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
 {
    fn eq(&self, other: &Self) -> bool {
        self.0.eq(&other.0) && self.1 == other.1
    }
 }

+#[cfg(feature = "diesel")]
+const _IMPL_DIESEL_NEW_TYPE_FOR_OBJECT_ID: () = {
+    use diesel::{
+        backend::Backend,
+        deserialize::{FromSql, FromStaticSqlRow},
+        expression::AsExpression,
+        internal::derives::as_expression::Bound,
+        pg::Pg,
+        query_builder::QueryId,
+        serialize,
+        serialize::{Output, ToSql},
+        sql_types::{HasSqlType, SingleValue, Text},
+        Expression,
+        Queryable,
+    };
+
+    // TODO: this impl only works for Postgres db because of to_string() call which requires reborrow
+    impl<Kind, ST> ToSql<ST, Pg> for ObjectId<Kind>
+    where
+        Kind: Object,
+        for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
+        String: ToSql<ST, Pg>,
+    {
+        fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result {
+            let v = self.0.to_string();
+            <String as ToSql<Text, Pg>>::to_sql(&v, &mut out.reborrow())
+        }
+    }
+    impl<'expr, Kind, ST> AsExpression<ST> for &'expr ObjectId<Kind>
+    where
+        Kind: Object,
+        for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
+        Bound<ST, String>: Expression<SqlType = ST>,
+        ST: SingleValue,
+    {
+        type Expression = Bound<ST, &'expr str>;
+        fn as_expression(self) -> Self::Expression {
+            Bound::new(self.0.as_str())
+        }
+    }
+    impl<Kind, ST> AsExpression<ST> for ObjectId<Kind>
+    where
+        Kind: Object,
+        for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
+        Bound<ST, String>: Expression<SqlType = ST>,
+        ST: SingleValue,
+    {
+        type Expression = Bound<ST, String>;
+        fn as_expression(self) -> Self::Expression {
+            Bound::new(self.0.to_string())
+        }
+    }
+    impl<Kind, ST, DB> FromSql<ST, DB> for ObjectId<Kind>
+    where
+        Kind: Object + Send + 'static,
+        for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
+        String: FromSql<ST, DB>,
+        DB: Backend,
+        DB: HasSqlType<ST>,
+    {
+        fn from_sql(
+            raw: DB::RawValue<'_>,
+        ) -> Result<Self, Box<dyn ::std::error::Error + Send + Sync>> {
+            let string: String = FromSql::<ST, DB>::from_sql(raw)?;
+            Ok(ObjectId::parse(&string)?)
+        }
+    }
+    impl<Kind, ST, DB> Queryable<ST, DB> for ObjectId<Kind>
+    where
+        Kind: Object + Send + 'static,
+        for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
+        String: FromStaticSqlRow<ST, DB>,
+        DB: Backend,
+        DB: HasSqlType<ST>,
+    {
+        type Row = String;
+        fn build(row: Self::Row) -> diesel::deserialize::Result<Self> {
+            Ok(ObjectId::parse(&row)?)
+        }
+    }
+    impl<Kind> QueryId for ObjectId<Kind>
+    where
+        Kind: Object + 'static,
+        for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
+    {
+        type QueryId = Self;
+    }
+};
+
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 pub mod tests {
    use super::*;
-    use crate::{fetch::object_id::should_refetch_object, traits::tests::DbUser};
+    use crate::traits::tests::DbUser;

    #[test]
    fn test_deserialize() {
@ -260,10 +362,10 @@ pub mod tests {

    #[test]
    fn test_should_refetch_object() {
-        let one_second_ago = Utc::now().naive_utc() - ChronoDuration::seconds(1);
+        let one_second_ago = Utc::now() - ChronoDuration::try_seconds(1).unwrap();
        assert!(!should_refetch_object(one_second_ago));

-        let two_days_ago = Utc::now().naive_utc() - ChronoDuration::days(2);
+        let two_days_ago = Utc::now() - ChronoDuration::try_days(2).unwrap();
        assert!(should_refetch_object(two_days_ago));
    }
 }
--- a/src/fetch/webfinger.rs
+++ b/src/fetch/webfinger.rs
@ -1,18 +1,42 @@
 use crate::{
    config::Data,
-    error::{Error, Error::WebfingerResolveFailed},
-    fetch::{fetch_object_http, object_id::ObjectId},
+    error::Error,
+    fetch::{fetch_object_http_with_accept, object_id::ObjectId},
    traits::{Actor, Object},
    FEDERATION_CONTENT_TYPE,
 };
-use anyhow::anyhow;
+use http::HeaderValue;
 use itertools::Itertools;
+use once_cell::sync::Lazy;
 use regex::Regex;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::{collections::HashMap, fmt::Display};
 use tracing::debug;
 use url::Url;

+/// Errors relative to webfinger handling
+#[derive(thiserror::Error, Debug)]
+pub enum WebFingerError {
+    /// The webfinger identifier is invalid
+    #[error("The webfinger identifier is invalid")]
+    WrongFormat,
+    /// The webfinger identifier doesn't match the expected instance domain name
+    #[error("The webfinger identifier doesn't match the expected instance domain name")]
+    WrongDomain,
+    /// The wefinger object did not contain any link to an activitypub item
+    #[error("The webfinger object did not contain any link to an activitypub item")]
+    NoValidLink,
+}
+
+impl WebFingerError {
+    fn into_crate_error(self) -> Error {
+        self.into()
+    }
+}
+
+/// The content-type for webfinger responses.
+pub static WEBFINGER_CONTENT_TYPE: HeaderValue = HeaderValue::from_static("application/jrd+json");
+
 /// Takes an identifier of the form `name@example.com`, and returns an object of `Kind`.
 ///
 /// For this the identifier is first resolved via webfinger protocol to an Activitypub ID. This ID
@ -24,19 +48,24 @@ pub async fn webfinger_resolve_actor<T: Clone, Kind>(
 where
    Kind: Object + Actor + Send + 'static + Object<DataType = T>,
    for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
-    <Kind as Object>::Error:
-        From<crate::error::Error> + From<anyhow::Error> + From<url::ParseError> + Send + Sync,
+    <Kind as Object>::Error: From<crate::error::Error> + Send + Sync + Display,
 {
    let (_, domain) = identifier
        .splitn(2, '@')
        .collect_tuple()
-        .ok_or(WebfingerResolveFailed)?;
+        .ok_or(WebFingerError::WrongFormat.into_crate_error())?;
    let protocol = if data.config.debug { "http" } else { "https" };
    let fetch_url =
        format!("{protocol}://{domain}/.well-known/webfinger?resource=acct:{identifier}");
    debug!("Fetching webfinger url: {}", &fetch_url);

-    let res: Webfinger = fetch_object_http(&Url::parse(&fetch_url)?, data).await?;
+    let res: Webfinger = fetch_object_http_with_accept(
+        &Url::parse(&fetch_url).map_err(Error::UrlParse)?,
+        data,
+        &WEBFINGER_CONTENT_TYPE,
+    )
+    .await?
+    .object;

    debug_assert_eq!(res.subject, format!("acct:{identifier}"));
    let links: Vec<Url> = res
@ -51,13 +80,15 @@ where
        })
        .filter_map(|l| l.href.clone())
        .collect();
+
    for l in links {
        let object = ObjectId::<Kind>::from(l).dereference(data).await;
-        if object.is_ok() {
-            return object;
+        match object {
+            Ok(obj) => return Ok(obj),
+            Err(error) => debug!(%error, "Failed to dereference link"),
        }
    }
-    Err(WebfingerResolveFailed.into())
+    Err(WebFingerError::NoValidLink.into_crate_error().into())
 }

 /// Extracts username from a webfinger resource parameter.
@ -66,24 +97,43 @@ where
 /// request. For a parameter of the form `acct:gargron@mastodon.social` it returns `gargron`.
 ///
 /// Returns an error if query doesn't match local domain.
-pub fn extract_webfinger_name<T>(query: &str, data: &Data<T>) -> Result<String, Error>
+///
+///```
+/// # use activitypub_federation::config::FederationConfig;
+/// # use activitypub_federation::traits::tests::DbConnection;
+/// # use activitypub_federation::fetch::webfinger::extract_webfinger_name;
+/// # tokio::runtime::Runtime::new().unwrap().block_on(async {
+/// # let db_connection = DbConnection;
+/// let config = FederationConfig::builder()
+///     .domain("example.com")
+///     .app_data(db_connection)
+///     .build()
+///     .await
+///     .unwrap();
+/// let data = config.to_request_data();
+/// let res = extract_webfinger_name("acct:test_user@example.com", &data).unwrap();
+/// assert_eq!(res, "test_user");
+/// # Ok::<(), anyhow::Error>(())
+/// }).unwrap();
+///```
+pub fn extract_webfinger_name<'i, T>(query: &'i str, data: &Data<T>) -> Result<&'i str, Error>
 where
    T: Clone,
 {
-    // TODO: would be nice if we could implement this without regex and remove the dependency
-    // Regex taken from Mastodon -
-    // https://github.com/mastodon/mastodon/blob/2b113764117c9ab98875141bcf1758ba8be58173/app/models/account.rb#L65
-    let regex = Regex::new(&format!(
-        "^acct:((?i)[a-z0-9_]+([a-z0-9_\\.-]+[a-z0-9_]+)?)@{}$",
-        data.domain()
-    ))
-    .map_err(Error::other)?;
-    Ok(regex
+    static WEBFINGER_REGEX: Lazy<Regex> =
+        Lazy::new(|| Regex::new(r"^acct:([\p{L}0-9_\.\-]+)@(.*)$").expect("compile regex"));
+    // Regex to extract usernames from webfinger query. Supports different alphabets using `\p{L}`.
+    // TODO: This should use a URL parser
+    let captures = WEBFINGER_REGEX
        .captures(query)
-        .and_then(|c| c.get(1))
-        .ok_or_else(|| Error::other(anyhow!("Webfinger regex failed to match")))?
-        .as_str()
-        .to_string())
+        .ok_or(WebFingerError::WrongFormat)?;
+
+    let account_name = captures.get(1).ok_or(WebFingerError::WrongFormat)?;
+
+    if captures.get(2).map(|m| m.as_str()) != Some(data.domain()) {
+        return Err(WebFingerError::WrongDomain.into());
+    }
+    Ok(account_name.as_str())
 }

 /// Builds a basic webfinger response for the actor.
@ -144,13 +194,14 @@ pub fn build_webfinger_response_with_type(
                    rel: Some("http://webfinger.net/rel/profile-page".to_string()),
                    kind: Some("text/html".to_string()),
                    href: Some(url.clone()),
-                    properties: Default::default(),
+                    ..Default::default()
                },
                WebfingerLink {
                    rel: Some("self".to_string()),
                    kind: Some(FEDERATION_CONTENT_TYPE.to_string()),
                    href: Some(url.clone()),
                    properties,
+                    ..Default::default()
                },
            ];
            acc.append(&mut links);
@ -186,12 +237,15 @@ pub struct WebfingerLink {
    pub kind: Option<String>,
    /// Url pointing to the target resource
    pub href: Option<Url>,
+    /// Used for remote follow external interaction url
+    pub template: Option<String>,
    /// Additional data about the link
    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
    pub properties: HashMap<Url, String>,
 }

 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod tests {
    use super::*;
    use crate::{
@ -200,7 +254,7 @@ mod tests {
    };

    #[tokio::test]
-    async fn test_webfinger() {
+    async fn test_webfinger() -> Result<(), Error> {
        let config = FederationConfig::builder()
            .domain("example.com")
            .app_data(DbConnection)
@ -208,9 +262,45 @@ mod tests {
            .await
            .unwrap();
        let data = config.to_request_data();
-        let res =
-            webfinger_resolve_actor::<DbConnection, DbUser>("LemmyDev@mastodon.social", &data)
-                .await;
-        assert!(res.is_ok());
+
+        webfinger_resolve_actor::<DbConnection, DbUser>("LemmyDev@mastodon.social", &data).await?;
+        // poa.st is as of 2023-07-14 the largest Pleroma instance
+        webfinger_resolve_actor::<DbConnection, DbUser>("graf@poa.st", &data).await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_webfinger_extract_name() -> Result<(), Error> {
+        use crate::traits::tests::DbConnection;
+        let data = Data {
+            config: FederationConfig::builder()
+                .domain("example.com")
+                .app_data(DbConnection)
+                .build()
+                .await
+                .unwrap(),
+            request_counter: Default::default(),
+        };
+        assert_eq!(
+            Ok("test123"),
+            extract_webfinger_name("acct:test123@example.com", &data)
+        );
+        assert_eq!(
+            Ok("Владимир"),
+            extract_webfinger_name("acct:Владимир@example.com", &data)
+        );
+        assert_eq!(
+            Ok("example.com"),
+            extract_webfinger_name("acct:example.com@example.com", &data)
+        );
+        assert_eq!(
+            Ok("da-sh"),
+            extract_webfinger_name("acct:da-sh@example.com", &data)
+        );
+        assert_eq!(
+            Ok("تجريب"),
+            extract_webfinger_name("acct:تجريب@example.com", &data)
+        );
+        Ok(())
    }
 }
--- a/src/http_signatures.rs
+++ b/src/http_signatures.rs
@ -1,7 +1,7 @@
 //! Generating keypairs, creating and verifying signatures
 //!
 //! Signature creation and verification is handled internally in the library. See
-//! [send_activity](crate::activity_queue::send_activity) and
+//! [send_activity](crate::activity_sending::SendActivityTask::sign_and_send) and
 //! [receive_activity (actix-web)](crate::actix_web::inbox::receive_activity) /
 //! [receive_activity (axum)](crate::axum::inbox::receive_activity).

@ -15,7 +15,10 @@ use crate::{
 use base64::{engine::general_purpose::STANDARD as Base64, Engine};
 use bytes::Bytes;
 use http::{header::HeaderName, uri::PathAndQuery, HeaderValue, Method, Uri};
-use http_signature_normalization_reqwest::prelude::{Config, SignExt};
+use http_signature_normalization_reqwest::{
+    prelude::{Config, SignExt},
+    DefaultSpawner,
+};
 use once_cell::sync::Lazy;
 use openssl::{
    hash::MessageDigest,
@ -67,9 +70,12 @@ pub fn generate_actor_keypair() -> Result<Keypair, std::io::Error> {
    })
 }

-/// Sets the amount of time that a signed request is valid.  Currenlty 5 minutes
-/// Mastodon & friends have ~1 hour expiry from creation if it's not set in the header
-pub(crate) const EXPIRES_AFTER: Duration = Duration::from_secs(300);
+/// Time for which HTTP signatures are valid.
+///
+/// This field is optional in the standard, but required by the Rust library. It is not clear
+/// what security concerns this expiration solves (if any), so we set a very high value of one hour
+/// to avoid any potential problems due to wrong clocks, overloaded servers or delayed delivery.
+pub(crate) const EXPIRES_AFTER: Duration = Duration::from_secs(60 * 60);

 /// Creates an HTTP post request to `inbox_url`, with the given `client` and `headers`, and
 /// `activity` as request body. The request is signed with `private_key` and then sent.
@ -79,8 +85,9 @@ pub(crate) async fn sign_request(
    activity: Bytes,
    private_key: PKey<Private>,
    http_signature_compat: bool,
-) -> Result<Request, anyhow::Error> {
-    static CONFIG: Lazy<Config> = Lazy::new(|| Config::new().set_expiration(EXPIRES_AFTER));
+) -> Result<Request, Error> {
+    static CONFIG: Lazy<Config<DefaultSpawner>> =
+        Lazy::new(|| Config::new().set_expiration(EXPIRES_AFTER));
    static CONFIG_COMPAT: Lazy<Config> = Lazy::new(|| {
        Config::new()
            .mastodon_compat()
@ -102,15 +109,12 @@ pub(crate) async fn sign_request(
                let mut signer = Signer::new(MessageDigest::sha256(), &private_key)?;
                signer.update(signing_string.as_bytes())?;

-                Ok(Base64.encode(signer.sign_to_vec()?)) as Result<_, anyhow::Error>
+                Ok(Base64.encode(signer.sign_to_vec()?)) as Result<_, Error>
            },
        )
        .await
 }

-static CONFIG2: Lazy<http_signature_normalization::Config> =
-    Lazy::new(http_signature_normalization::Config::new);
-
 /// Verifies the HTTP signature on an incoming federation request
 /// for a given actor's public key.
 ///
@ -147,7 +151,7 @@ pub(crate) async fn signing_actor<'a, A, H>(
 ) -> Result<A, <A as Object>::Error>
 where
    A: Object + Actor,
-    <A as Object>::Error: From<Error> + From<anyhow::Error>,
+    <A as Object>::Error: From<Error>,
    for<'de2> <A as Object>::Kind: Deserialize<'de2>,
    H: IntoIterator<Item = (&'a HeaderName, &'a HeaderValue)>,
 {
@ -185,12 +189,18 @@ fn verify_signature_inner(
    uri: &Uri,
    public_key: &str,
 ) -> Result<(), Error> {
+    static CONFIG: Lazy<http_signature_normalization::Config> = Lazy::new(|| {
+        http_signature_normalization::Config::new()
+            .set_expiration(EXPIRES_AFTER)
+            .require_digest()
+    });
+
    let path_and_query = uri.path_and_query().map(PathAndQuery::as_str).unwrap_or("");

-    let verified = CONFIG2
+    let verified = CONFIG
        .begin_verify(method.as_str(), path_and_query, header_map)
-        .map_err(Error::other)?
-        .verify(|signature, signing_string| -> anyhow::Result<bool> {
+        .map_err(|val| Error::Other(val.to_string()))?
+        .verify(|signature, signing_string| -> Result<bool, Error> {
            debug!(
                "Verifying with key {}, message {}",
                &public_key, &signing_string
@ -198,9 +208,13 @@ fn verify_signature_inner(
            let public_key = PKey::public_key_from_pem(public_key.as_bytes())?;
            let mut verifier = Verifier::new(MessageDigest::sha256(), &public_key)?;
            verifier.update(signing_string.as_bytes())?;
-            Ok(verifier.verify(&Base64.decode(signature)?)?)
-        })
-        .map_err(Error::other)?;
+
+            let base64_decoded = Base64
+                .decode(signature)
+                .map_err(|err| Error::Other(err.to_string()))?;
+
+            Ok(verifier.verify(&base64_decoded)?)
+        })?;

    if verified {
        debug!("verified signature for {}", uri);
@ -264,9 +278,10 @@ pub(crate) fn verify_body_hash(
 }

 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 pub mod test {
    use super::*;
-    use crate::activity_queue::generate_request_headers;
+    use crate::activity_sending::generate_request_headers;
    use reqwest::Client;
    use reqwest_middleware::ClientWithMiddleware;
    use std::str::FromStr;
--- a/src/lib.rs
+++ b/src/lib.rs
@ -11,6 +11,7 @@
 #![deny(missing_docs)]

 pub mod activity_queue;
+pub mod activity_sending;
 #[cfg(feature = "actix-web")]
 pub mod actix_web;
 #[cfg(feature = "axum")]
@ -23,7 +24,51 @@ pub mod protocol;
 pub(crate) mod reqwest_shim;
 pub mod traits;

+use crate::{
+    config::Data,
+    error::Error,
+    fetch::object_id::ObjectId,
+    traits::{ActivityHandler, Actor, Object},
+};
 pub use activitystreams_kinds as kinds;

+use serde::{de::DeserializeOwned, Deserialize};
+use url::Url;
+
 /// Mime type for Activitypub data, used for `Accept` and `Content-Type` HTTP headers
-pub static FEDERATION_CONTENT_TYPE: &str = "application/activity+json";
+pub const FEDERATION_CONTENT_TYPE: &str = "application/activity+json";
+
+/// Deserialize incoming inbox activity to the given type, perform basic
+/// validation and extract the actor.
+async fn parse_received_activity<Activity, ActorT, Datatype>(
+    body: &[u8],
+    data: &Data<Datatype>,
+) -> Result<(Activity, ActorT), <Activity as ActivityHandler>::Error>
+where
+    Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
+    ActorT: Object<DataType = Datatype> + Actor + Send + 'static,
+    for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
+    <Activity as ActivityHandler>::Error: From<Error> + From<<ActorT as Object>::Error>,
+    <ActorT as Object>::Error: From<Error>,
+    Datatype: Clone,
+{
+    let activity: Activity = serde_json::from_slice(body).map_err(|e| {
+        // Attempt to include activity id in error message
+        let id = extract_id(body).ok();
+        Error::ParseReceivedActivity(e, id)
+    })?;
+    data.config.verify_url_and_domain(&activity).await?;
+    let actor = ObjectId::<ActorT>::from(activity.actor().clone())
+        .dereference(data)
+        .await?;
+    Ok((activity, actor))
+}
+
+/// Attempt to parse id field from serialized json
+fn extract_id(data: &[u8]) -> serde_json::Result<Url> {
+    #[derive(Deserialize)]
+    struct Id {
+        id: Url,
+    }
+    Ok(serde_json::from_slice::<Id>(data)?.id)
+}
--- a/src/protocol/context.rs
+++ b/src/protocol/context.rs
@ -15,11 +15,11 @@
 //! };
 //! let note_with_context = WithContext::new_default(note);
 //! let serialized = serde_json::to_string(&note_with_context)?;
-//! assert_eq!(serialized, r#"{"@context":["https://www.w3.org/ns/activitystreams"],"content":"Hello world"}"#);
+//! assert_eq!(serialized, r#"{"@context":"https://www.w3.org/ns/activitystreams","content":"Hello world"}"#);
 //! Ok::<(), serde_json::error::Error>(())
 //! ```

-use crate::{config::Data, protocol::helpers::deserialize_one_or_many, traits::ActivityHandler};
+use crate::{config::Data, traits::ActivityHandler};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use url::Url;
@ -31,8 +31,7 @@ const DEFAULT_CONTEXT: &str = "https://www.w3.org/ns/activitystreams";
 #[derive(Serialize, Deserialize, Debug)]
 pub struct WithContext<T> {
    #[serde(rename = "@context")]
-    #[serde(deserialize_with = "deserialize_one_or_many")]
-    context: Vec<Value>,
+    context: Value,
    #[serde(flatten)]
    inner: T,
 }
@ -40,12 +39,12 @@ pub struct WithContext<T> {
 impl<T> WithContext<T> {
    /// Create a new wrapper with the default Activitypub context.
    pub fn new_default(inner: T) -> WithContext<T> {
-        let context = vec![Value::String(DEFAULT_CONTEXT.to_string())];
+        let context = Value::String(DEFAULT_CONTEXT.to_string());
        WithContext::new(inner, context)
    }

    /// Create new wrapper with custom context. Use this in case you are implementing extensions.
-    pub fn new(inner: T, context: Vec<Value>) -> WithContext<T> {
+    pub fn new(inner: T, context: Value) -> WithContext<T> {
        WithContext { context, inner }
    }

--- a/src/protocol/helpers.rs
+++ b/src/protocol/helpers.rs
@ -56,12 +56,12 @@ where
 /// #[derive(serde::Deserialize)]
 /// struct Note {
 ///     #[serde(deserialize_with = "deserialize_one")]
-///     to: Url
+///     to: [Url; 1]
 /// }
 ///
 /// let note = serde_json::from_str::<Note>(r#"{"to": ["https://example.com/u/alice"] }"#);
 /// assert!(note.is_ok());
-pub fn deserialize_one<'de, T, D>(deserializer: D) -> Result<T, D::Error>
+pub fn deserialize_one<'de, T, D>(deserializer: D) -> Result<[T; 1], D::Error>
 where
    T: Deserialize<'de>,
    D: Deserializer<'de>,
@ -75,8 +75,8 @@ where

    let result: MaybeArray<T> = Deserialize::deserialize(deserializer)?;
    Ok(match result {
-        MaybeArray::Simple(value) => value,
-        MaybeArray::Array([value]) => value,
+        MaybeArray::Simple(value) => [value],
+        MaybeArray::Array([value]) => [value],
    })
 }

@ -115,3 +115,22 @@ where
    let inner = T::deserialize(value).unwrap_or_default();
    Ok(inner)
 }
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn deserialize_one_multiple_values() {
+        use crate::protocol::helpers::deserialize_one;
+        use url::Url;
+        #[derive(serde::Deserialize)]
+        struct Note {
+            #[serde(deserialize_with = "deserialize_one")]
+            _to: [Url; 1],
+        }
+
+        let note = serde_json::from_str::<Note>(
+            r#"{"_to": ["https://example.com/u/alice", "https://example.com/u/bob"] }"#,
+        );
+        assert!(note.is_err());
+    }
+}
--- a/src/protocol/public_key.rs
+++ b/src/protocol/public_key.rs
@ -6,7 +6,7 @@ use url::Url;
 /// Public key of actors which is used for HTTP signatures.
 ///
 /// This needs to be federated in the `public_key` field of all actors.
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct PublicKey {
    /// Id of this private key.
--- a/src/protocol/values.rs
+++ b/src/protocol/values.rs
@ -35,7 +35,7 @@ use serde::{Deserialize, Serialize};
 /// Media type for markdown text.
 ///
 /// <https://www.iana.org/assignments/media-types/media-types.xhtml>
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
 pub enum MediaTypeMarkdown {
    /// `text/markdown`
    #[serde(rename = "text/markdown")]
@ -45,7 +45,7 @@ pub enum MediaTypeMarkdown {
 /// Media type for HTML text.
 ///
 /// <https://www.iana.org/assignments/media-types/media-types.xhtml>
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
 pub enum MediaTypeHtml {
    /// `text/html`
    #[serde(rename = "text/html")]
--- a/src/reqwest_shim.rs
+++ b/src/reqwest_shim.rs
@ -3,17 +3,15 @@ use bytes::{BufMut, Bytes, BytesMut};
 use futures_core::{ready, stream::BoxStream, Stream};
 use pin_project_lite::pin_project;
 use reqwest::Response;
-use serde::de::DeserializeOwned;
 use std::{
    future::Future,
-    marker::PhantomData,
    mem,
    pin::Pin,
    task::{Context, Poll},
 };

-/// 100KB
-const MAX_BODY_SIZE: usize = 102400;
+/// 200KB
+const MAX_BODY_SIZE: usize = 204800;

 pin_project! {
    pub struct BytesFuture {
@ -30,10 +28,7 @@ impl Future for BytesFuture {
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        loop {
            let this = self.as_mut().project();
-            if let Some(chunk) = ready!(this.stream.poll_next(cx))
-                .transpose()
-                .map_err(Error::other)?
-            {
+            if let Some(chunk) = ready!(this.stream.poll_next(cx)).transpose()? {
                this.aggregator.put(chunk);
                if this.aggregator.len() > *this.limit {
                    return Poll::Ready(Err(Error::ResponseBodyLimit));
@ -49,27 +44,6 @@ impl Future for BytesFuture {
    }
 }

-pin_project! {
-    pub struct JsonFuture<T> {
-        _t: PhantomData<T>,
-        #[pin]
-        future: BytesFuture,
-    }
-}
-
-impl<T> Future for JsonFuture<T>
-where
-    T: DeserializeOwned,
-{
-    type Output = Result<T, Error>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.project();
-        let bytes = ready!(this.future.poll(cx))?;
-        Poll::Ready(serde_json::from_slice(&bytes).map_err(Error::other))
-    }
-}
-
 pin_project! {
    pub struct TextFuture {
        #[pin]
@ -83,7 +57,7 @@ impl Future for TextFuture {
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        let bytes = ready!(this.future.poll(cx))?;
-        Poll::Ready(String::from_utf8(bytes.to_vec()).map_err(Error::other))
+        Poll::Ready(String::from_utf8(bytes.to_vec()).map_err(Error::Utf8))
    }
 }

@ -97,20 +71,16 @@ impl Future for TextFuture {
 /// TODO: Remove this shim as soon as reqwest gets support for size-limited bodies.
 pub trait ResponseExt {
    type BytesFuture;
-    type JsonFuture<T>;
    type TextFuture;

    /// Size limited version of `bytes` to work around a reqwest issue. Check [`ResponseExt`] docs for details.
    fn bytes_limited(self) -> Self::BytesFuture;
-    /// Size limited version of `json` to work around a reqwest issue. Check [`ResponseExt`] docs for details.
-    fn json_limited<T>(self) -> Self::JsonFuture<T>;
    /// Size limited version of `text` to work around a reqwest issue. Check [`ResponseExt`] docs for details.
    fn text_limited(self) -> Self::TextFuture;
 }

 impl ResponseExt for Response {
    type BytesFuture = BytesFuture;
-    type JsonFuture<T> = JsonFuture<T>;
    type TextFuture = TextFuture;

    fn bytes_limited(self) -> Self::BytesFuture {
@ -121,13 +91,6 @@ impl ResponseExt for Response {
        }
    }

-    fn json_limited<T>(self) -> Self::JsonFuture<T> {
-        JsonFuture {
-            _t: PhantomData,
-            future: self.bytes_limited(),
-        }
-    }
-
    fn text_limited(self) -> Self::TextFuture {
        TextFuture {
            future: self.bytes_limited(),
--- a/src/traits.rs
+++ b/src/traits.rs
@ -2,7 +2,7 @@

 use crate::{config::Data, protocol::public_key::PublicKey};
 use async_trait::async_trait;
-use chrono::NaiveDateTime;
+use chrono::{DateTime, Utc};
 use serde::Deserialize;
 use std::{fmt::Debug, ops::Deref};
 use url::Url;
@ -11,7 +11,7 @@ use url::Url;
 ///
 /// ```
 /// # use activitystreams_kinds::{object::NoteType, public};
-/// # use chrono::{Local, NaiveDateTime};
+/// # use chrono::{Local, DateTime, Utc};
 /// # use serde::{Deserialize, Serialize};
 /// # use url::Url;
 /// # use activitypub_federation::protocol::{public_key::PublicKey, helpers::deserialize_one_or_many};
@ -22,6 +22,7 @@ use url::Url;
 /// # use activitypub_federation::traits::tests::{DbConnection, DbUser};
 /// #
 /// /// How the post is read/written in the local database
+/// #[derive(Debug)]
 /// pub struct DbPost {
 ///     pub text: String,
 ///     pub ap_id: ObjectId<DbPost>,
@ -93,7 +94,7 @@ use url::Url;
 ///
 /// }
 #[async_trait]
-pub trait Object: Sized {
+pub trait Object: Sized + Debug {
    /// App data type passed to handlers. Must be identical to
    /// [crate::config::FederationConfigBuilder::app_data] type.
    type DataType: Clone + Send + Sync;
@ -111,7 +112,7 @@ pub trait Object: Sized {
    ///
    /// The object is refetched if `last_refreshed_at` value is more than 24 hours ago. In debug
    /// mode this is reduced to 20 seconds.
-    fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    fn last_refreshed_at(&self) -> Option<DateTime<Utc>> {
        None
    }

@ -339,12 +340,12 @@ pub trait Collection: Sized {
 pub mod tests {
    use super::*;
    use crate::{
+        error::Error,
        fetch::object_id::ObjectId,
        http_signatures::{generate_actor_keypair, Keypair},
-        protocol::{public_key::PublicKey, verification::verify_domains_match},
+        protocol::verification::verify_domains_match,
    };
    use activitystreams_kinds::{activity::FollowType, actor::PersonType};
-    use anyhow::Error;
    use once_cell::sync::Lazy;
    use serde::{Deserialize, Serialize};

@ -355,7 +356,7 @@ pub mod tests {
        pub async fn read_post_from_json_id<T>(&self, _: Url) -> Result<Option<T>, Error> {
            Ok(None)
        }
-        pub async fn read_local_user(&self, _: String) -> Result<DbUser, Error> {
+        pub async fn read_local_user(&self, _: &str) -> Result<DbUser, Error> {
            todo!()
        }
        pub async fn upsert<T>(&self, _: &T) -> Result<(), Error> {
Author	SHA1	Message	Date
Felix Ableitner	16844f048a	Version 0.5.6	2024-05-06 11:09:47 +02:00
Nutomic	cf1f84993b	Make response content-type check case insensitive (#111 ) * Make response content-type check case insensitive For wordpress compat * cleaner * clippy * fmt * fmt	2024-05-06 11:09:23 +02:00
Felix Ableitner	24afad7abc	Version 0.5.5	2024-05-02 13:06:22 +02:00
Nutomic	c48de9e944	Upgrade dependencies (#110 )	2024-05-02 06:58:08 -04:00
Nutomic	be69efdee3	Require signed digest when verifying signatures (#109 )	2024-05-02 10:58:56 +02:00
Nutomic	ddc455510b	Dont crash when calling is_local_url() without domain (#108 )	2024-05-02 10:58:33 +02:00
Felix Ableitner	ee268405f7	Version 0.5.4	2024-04-10 11:32:14 +02:00
Nutomic	54e8a1145f	Add function ObjectId.is_local (#106 ) * Add function ObjectId.is_local * add test * add test	2024-04-10 11:31:55 +02:00
Felix Ableitner	779313ac22	Version 0.5.3	2024-04-09 11:30:43 +02:00
Nutomic	7def01a19a	Avoid running ci checks twice (#105 ) * Avoid running ci checks twice * upgrade rust * move clippy config to cargo.toml	2024-04-09 11:28:57 +02:00
Nutomic	a2ac97db98	Allow fetching from local domain in case it redirects to remote (#104 ) * Allow fetching from local domain in case it redirects to remote * clippy * fix lemmy tests	2024-04-09 11:28:22 +02:00
Nutomic	5402bc9c19	Retry activity send in case of timeout or rate limit (#102 )	2024-04-09 10:38:08 +02:00
Felix Ableitner	1b46dd6f80	Version 0.5.2	2024-03-08 15:43:43 +01:00
Nutomic	da28c9c890	Fix request_counter check because fetch_add returns original value (#97 )	2024-03-08 09:36:44 -05:00
Nutomic	9b31a7b44b	Get Mastodon signed fetch working (#98 ) * debug signed fetch * regex * content-type * no dbg * clippy	2024-03-08 15:26:57 +01:00
Felix Ableitner	147f144769	Version 0.5.1	2024-03-04 16:27:44 +01:00
Felix Ableitner	3b5e5f66ba	Upgrade dependencies	2024-03-04 16:23:46 +01:00
Nutomic	636b47c8b2	Add back activity send queue as optional feature (#94 ) * Add back activity send queue as optional feature * fix port collision in tests * improve docs * serialize fn * deduplicate * more dedup * more dedup * dedupupup * test cleanup * remove fn	2024-03-04 08:53:33 -05:00
Nutomic	a859db05bb	Add security checks when fetching objects (#95 ) * Add security checks when fetching objects * static * as ref * update comment * fix header * Update comment	2024-03-04 12:47:42 +01:00
Felix Ableitner	f907b6efa7	Version 0.5.1-beta.1	2024-01-05 16:21:20 +01:00
Nutomic	ec97b44de4	Fix return type for deserialize_one (#93 )	2024-01-05 10:18:12 -05:00
Felix Ableitner	3efa99514c	Version 0.5.0	2024-01-02 11:40:55 +01:00
Felix Ableitner	9c3c756890	Version 0.5.0-beta.7	2023-12-20 12:28:43 +01:00
Nutomic	9e8d466b40	Better JSON error messages (#91 )	2023-12-20 06:19:58 -05:00
Nutomic	709f29b7f8	Context doesnt have to be an array (#90 )	2023-12-20 11:21:33 +01:00
Felix Ableitner	fec0af2406	Version 0.5.0-beta.6	2023-12-12 11:42:52 +01:00
Felix Ableitner	71ece55641	Upgrade deps	2023-12-12 11:32:22 +01:00
Nutomic	50db596ce0	Better error when activity receive fails (#89 ) * Minor refactoring * Better error when receive fails * clippy * add test case * comments * take ref	2023-12-12 11:30:21 +01:00
Soso	12aad8bf3c	Webfinger: don't discard consumer errors (#85 ) * Improve WebFinger errors * Improve webfinger extraction * Fix typo * Document webfinger parsing * Reimplement Regex based webfinger parsing * clippy * no unwrap --------- Co-authored-by: Felix Ableitner <me@nutomic.com>	2023-12-11 22:48:32 +01:00
Nutomic	24830070f6	Add diesel feature, add ObjectId::dereference_forced (#88 ) * Add diesel feature This can simplify Lemmy code and avoid converting back and forth to DbUrl type all the time. * Also add diesel derives for CollectionId * Add ObjectId::dereference_forced * no deprecated code * fmt	2023-12-11 15:04:18 +01:00
Nutomic	1f7de85a53	Upgrade dependencies (#86 )	2023-12-01 15:40:12 +01:00
Nutomic	69b80aa6e1	Change impl of ObjectId::parse (#84 ) * Change impl of ObjectId::parse It should be consistent with Url::parse * fmt	2023-11-24 11:21:00 +01:00
Felix Ableitner	33649b43b7	Version 0.5.0-beta.5	2023-11-20 14:35:51 +01:00
cetra3	098a4299f0	Remove `anyhow` from trait definitions (#82 )	2023-11-20 11:42:47 +01:00
Nutomic	679228873a	Implement PartialEq for testing (#81 )	2023-11-13 10:38:58 +01:00
Felix Ableitner	171d32720e	0.5.0-beta.4	2023-10-24 11:34:42 +02:00
Nutomic	e86330852d	Support different alphabets in webfinger username (#78 ) * Support different alphabets in webfinger username * clippy	2023-10-24 11:08:55 +02:00
Nutomic	ec12fb3830	Support fetches with redirect (#76 ) * Support fetches with redirect * pub	2023-10-24 10:53:59 +02:00
Felix Ableitner	a5102d0633	Version 0.5.0-beta.3	2023-09-01 11:20:36 +02:00
Felix Ableitner	99e2226993	Version 0.5.0-beta.2	2023-09-01 11:20:07 +02:00
phiresky	51443aa57c	Remove activity queue and add raw sending (#75 ) * make prepare_raw, sign_raw, send_raw functions public * remove in-memory activity queue * rename module * comment * don"t clone * fix doc comment * remove send_activity function --------- Co-authored-by: Nutomic <me@nutomic.com>	2023-09-01 11:19:22 +02:00
Nutomic	9477180b4e	Add webfinger template field, used for remote follow (#74 ) https://socialhub.activitypub.rocks/t/what-is-the-current-spec-for-remote-follow/2020/15	2023-08-31 13:42:58 +02:00
Felix Ableitner	b0547e7793	Test that deserialize_one errors on multiple array values	2023-08-09 11:12:30 +02:00
Felix Ableitner	7bb17f21d5	0.5.0-beta.1	2023-07-26 16:27:33 +02:00
Nutomic	426871f5af	Use anyhow::Error for UrlVerifier return type (fixes #61 ) (#65 ) * Use anyhow::Error for UrlVerifier return type (fixes #61) * fmt * uncomment	2023-07-26 10:26:22 -04:00
phiresky	32e3cd5574	make time-zone aware (#62 ) * make time-zone aware. * format and revert debug * better test log * empty	2023-07-26 16:17:39 +02:00
Nutomic	61085a643f	Fix woodpecker badge in readme	2023-07-20 16:39:22 +02:00
Felix Ableitner	9b5d6af8c0	Version 0.4.6	2023-07-20 16:37:20 +02:00
Nutomic	b63445afca	Update reqwest_shim.rs (#72 )	2023-07-20 10:27:51 -04:00
Samuel Tardieu	02ab897f4f	Fix webfinger fetching non-compliance (#69 ) Allow webfinger to use "jrd+json" instead of "activity+json". Fix #68.	2023-07-18 16:17:21 -04:00
Nutomic	988450c79f	Fix tests (#71 )	2023-07-18 16:10:25 -04:00
phiresky	af92e0d532	add shutdown method (#53 ) * add shutdown method * simplify shutdown interface * make work on rust < 1.70 * upgrade ci to rust 1.70 * make clippy and linux torvalds happy	2023-07-04 16:08:39 +02:00
phiresky	68f9210d4c	add a separate allow_http flag (#54 ) Co-authored-by: Nutomic <me@nutomic.com>	2023-07-03 16:24:11 +02:00
Colin Atkinson	d9f1a4414f	Fix regex error when actix-web feature not enabled (#56 ) * Fix formatting for nightly rustfmt https://github.com/LemmyNet/lemmy/issues/3467 * Fix regex error when actix-web feature not enabled If the crate is built with only the axum feature, compiling the webfinger account regex will fail with an error "Unicode-aware case insensitivity matching is not available..." because of the missing unicode-case feature. This doesn't happen if actix is installed because it pulls in the regex crate with all features (via [actix-router][0]). The failure can be demonstrated by reverting this commit's change to Cargo.toml and running: cargo test --no-default-features --features=axum --doc extract_webfinger_name Resolve this by adding the unicode-case feature to the regex dependency. [0]: `0e8ed50e3a/actix-router/Cargo.toml (L25)`	2023-07-03 15:05:18 +02:00
phiresky	b64f4a8f3f	fix: make "other" error actually transparent (#51 ) * fix: make "other" error actually transparent * cargo fmt	2023-06-29 10:19:49 +02:00
Felix Ableitner	93b7aa7979	Version 0.4.5	2023-06-27 15:42:50 +02:00
Nutomic	325f66ba32	Fix HTTP signature expiration (ref #46 ) (#52 )	2023-06-27 09:46:41 +02:00
Felix Ableitner	7300940e10	Version 0.4.4	2023-06-22 09:42:27 +02:00
cetra3	5de4a34550	Add a no limit option to the config (#45 ) * Add a no limit option to the config * Set defaults to `0`	2023-06-22 09:40:59 +02:00
Felix Ableitner	607aca7739	Version 0.4.3	2023-06-22 09:21:27 +02:00
Nutomic	cfcde0dcc4	Retry activity send on connection failure (fixes #41 ) (#48 )	2023-06-22 09:21:06 +02:00
Nutomic	3d9d54cf09	Increase HTTP signature expiration time to one day (fixes #46 ) (#47 )	2023-06-22 09:20:57 +02:00
Peter de Witte	8f997ec340	Adding security-considerations to 02_overview.md (#44 ) * Adding security-considerations to 02_overview.md * Updated layout	2023-06-20 13:48:32 +02:00