crates/apub_lib/src/object_id.rs

   1 use crate::{traits::ApubObject, APUB_JSON_CONTENT_TYPE};
   2 use activitystreams::chrono::{Duration as ChronoDuration, NaiveDateTime, Utc};
   3 use anyhow::anyhow;
   4 use diesel::NotFound;
   5 use lemmy_utils::{
   6   request::{build_user_agent, retry},
   7   settings::structs::Settings,
   8   LemmyError,
   9 };
  10 use log::info;
  11 use once_cell::sync::Lazy;
  12 use reqwest::{Client, StatusCode};
  13 use serde::{Deserialize, Serialize};
  14 use std::{
  15   fmt::{Debug, Display, Formatter},
  16   marker::PhantomData,
  17   time::Duration,
  18 };
  19 use url::Url;
  20
  21 /// Maximum number of HTTP requests allowed to handle a single incoming activity (or a single object
  22 /// fetch through the search). This should be configurable.
  23 static REQUEST_LIMIT: i32 = 25;
  24
  25 static CLIENT: Lazy<Client> = Lazy::new(|| {
  26   Client::builder()
  27     .user_agent(build_user_agent(&Settings::get()))
  28     .build()
  29     .expect("Couldn't build client")
  30 });
  31
  32 /// We store Url on the heap because it is quite large (88 bytes).
  33 #[derive(Clone, PartialEq, Serialize, Deserialize, Debug)]
  34 #[serde(transparent)]
  35 pub struct ObjectId<Kind>(Box<Url>, #[serde(skip)] PhantomData<Kind>)
  36 where
  37   Kind: ApubObject + Send + 'static,
  38   for<'de2> <Kind as ApubObject>::ApubType: serde::Deserialize<'de2>;
  39
  40 impl<Kind> ObjectId<Kind>
  41 where
  42   Kind: ApubObject + Send + 'static,
  43   for<'de2> <Kind as ApubObject>::ApubType: serde::Deserialize<'de2>,
  44 {
  45   pub fn new<T>(url: T) -> Self
  46   where
  47     T: Into<Url>,
  48   {
  49     ObjectId(Box::new(url.into()), PhantomData::<Kind>)
  50   }
  51
  52   pub fn inner(&self) -> &Url {
  53     &self.0
  54   }
  55
  56   /// Fetches an activitypub object, either from local database (if possible), or over http.
  57   pub async fn dereference(
  58     &self,
  59     data: &<Kind as ApubObject>::DataType,
  60     request_counter: &mut i32,
  61   ) -> Result<Kind, LemmyError> {
  62     let db_object = self.dereference_from_db(data).await?;
  63
  64     // if its a local object, only fetch it from the database and not over http
  65     if self.0.domain() == Some(&Settings::get().get_hostname_without_port()?) {
  66       return match db_object {
  67         None => Err(NotFound {}.into()),
  68         Some(o) => Ok(o),
  69       };
  70     }
  71
  72     // object found in database
  73     if let Some(object) = db_object {
  74       // object is old and should be refetched
  75       if let Some(last_refreshed_at) = object.last_refreshed_at() {
  76         if should_refetch_object(last_refreshed_at) {
  77           return self
  78             .dereference_from_http(data, request_counter, Some(object))
  79             .await;
  80         }
  81       }
  82       Ok(object)
  83     }
  84     // object not found, need to fetch over http
  85     else {
  86       self
  87         .dereference_from_http(data, request_counter, None)
  88         .await
  89     }
  90   }
  91
  92   /// Fetch an object from the local db. Instead of falling back to http, this throws an error if
  93   /// the object is not found in the database.
  94   pub async fn dereference_local(
  95     &self,
  96     data: &<Kind as ApubObject>::DataType,
  97   ) -> Result<Kind, LemmyError> {
  98     let object = self.dereference_from_db(data).await?;
  99     object.ok_or_else(|| anyhow!("object not found in database {}", self).into())
 100   }
 101
 102   /// returning none means the object was not found in local db
 103   async fn dereference_from_db(
 104     &self,
 105     data: &<Kind as ApubObject>::DataType,
 106   ) -> Result<Option<Kind>, LemmyError> {
 107     let id = self.0.clone();
 108     ApubObject::read_from_apub_id(*id, data).await
 109   }
 110
 111   async fn dereference_from_http(
 112     &self,
 113     data: &<Kind as ApubObject>::DataType,
 114     request_counter: &mut i32,
 115     db_object: Option<Kind>,
 116   ) -> Result<Kind, LemmyError> {
 117     // dont fetch local objects this way
 118     debug_assert!(self.0.domain() != Some(&Settings::get().hostname));
 119     info!("Fetching remote object {}", self.to_string());
 120
 121     *request_counter += 1;
 122     if *request_counter > REQUEST_LIMIT {
 123       return Err(LemmyError::from(anyhow!("Request limit reached")));
 124     }
 125
 126     let res = retry(|| {
 127       CLIENT
 128         .get(self.0.as_str())
 129         .header("Accept", APUB_JSON_CONTENT_TYPE)
 130         .timeout(Duration::from_secs(60))
 131         .send()
 132     })
 133     .await?;
 134
 135     if res.status() == StatusCode::GONE {
 136       if let Some(db_object) = db_object {
 137         db_object.delete(data).await?;
 138       }
 139       return Err(anyhow!("Fetched remote object {} which was deleted", self).into());
 140     }
 141
 142     let res2: Kind::ApubType = res.json().await?;
 143
 144     Kind::verify(&res2, self.inner(), data, request_counter).await?;
 145     Ok(Kind::from_apub(res2, data, request_counter).await?)
 146   }
 147 }
 148
 149 static ACTOR_REFETCH_INTERVAL_SECONDS: i64 = 24 * 60 * 60;
 150 static ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG: i64 = 10;
 151
 152 /// Determines when a remote actor should be refetched from its instance. In release builds, this is
 153 /// `ACTOR_REFETCH_INTERVAL_SECONDS` after the last refetch, in debug builds
 154 /// `ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG`.
 155 ///
 156 /// TODO it won't pick up new avatars, summaries etc until a day after.
 157 /// Actors need an "update" activity pushed to other servers to fix this.
 158 fn should_refetch_object(last_refreshed: NaiveDateTime) -> bool {
 159   let update_interval = if cfg!(debug_assertions) {
 160     // avoid infinite loop when fetching community outbox
 161     ChronoDuration::seconds(ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG)
 162   } else {
 163     ChronoDuration::seconds(ACTOR_REFETCH_INTERVAL_SECONDS)
 164   };
 165   let refresh_limit = Utc::now().naive_utc() - update_interval;
 166   last_refreshed.lt(&refresh_limit)
 167 }
 168
 169 impl<Kind> Display for ObjectId<Kind>
 170 where
 171   Kind: ApubObject + Send + 'static,
 172   for<'de2> <Kind as ApubObject>::ApubType: serde::Deserialize<'de2>,
 173 {
 174   #[allow(clippy::to_string_in_display)]
 175   fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
 176     // Use to_string here because Url.display is not useful for us
 177     write!(f, "{}", self.0.to_string())
 178   }
 179 }
 180
 181 impl<Kind> From<ObjectId<Kind>> for Url
 182 where
 183   Kind: ApubObject + Send + 'static,
 184   for<'de2> <Kind as ApubObject>::ApubType: serde::Deserialize<'de2>,
 185 {
 186   fn from(id: ObjectId<Kind>) -> Self {
 187     *id.0
 188   }
 189 }
 190
 191 #[cfg(test)]
 192 mod tests {
 193   use super::*;
 194   use crate::object_id::should_refetch_object;
 195
 196   #[test]
 197   fn test_should_refetch_object() {
 198     let one_second_ago = Utc::now().naive_utc() - ChronoDuration::seconds(1);
 199     assert!(!should_refetch_object(one_second_ago));
 200
 201     let two_days_ago = Utc::now().naive_utc() - ChronoDuration::days(2);
 202     assert!(should_refetch_object(two_days_ago));
 203   }
 204 }