server/src/apub/fetcher.rs

   1 use activitystreams::object::Note;
   2 use actix_web::Result;
   3 use diesel::{result::Error::NotFound, PgConnection};
   4 use failure::{Error, _core::fmt::Debug};
   5 use isahc::prelude::*;
   6 use log::debug;
   7 use serde::Deserialize;
   8 use std::time::Duration;
   9 use url::Url;
  10
  11 use crate::{
  12   api::site::SearchResponse,
  13   db::{
  14     comment::{Comment, CommentForm},
  15     comment_view::CommentView,
  16     community::{Community, CommunityForm, CommunityModerator, CommunityModeratorForm},
  17     community_view::CommunityView,
  18     post::{Post, PostForm},
  19     post_view::PostView,
  20     user::{UserForm, User_},
  21     Crud,
  22     Joinable,
  23     SearchType,
  24   },
  25   naive_now,
  26   routes::nodeinfo::{NodeInfo, NodeInfoWellKnown},
  27 };
  28
  29 use crate::{
  30   apub::{
  31     get_apub_protocol_string,
  32     is_apub_id_valid,
  33     FromApub,
  34     GroupExt,
  35     PageExt,
  36     PersonExt,
  37     APUB_JSON_CONTENT_TYPE,
  38   },
  39   db::user_view::UserView,
  40 };
  41 use chrono::NaiveDateTime;
  42
  43 static ACTOR_REFETCH_INTERVAL_SECONDS: i64 = 24 * 60 * 60;
  44
  45 // Fetch nodeinfo metadata from a remote instance.
  46 fn _fetch_node_info(domain: &str) -> Result<NodeInfo, Error> {
  47   let well_known_uri = Url::parse(&format!(
  48     "{}://{}/.well-known/nodeinfo",
  49     get_apub_protocol_string(),
  50     domain
  51   ))?;
  52   let well_known = fetch_remote_object::<NodeInfoWellKnown>(&well_known_uri)?;
  53   Ok(fetch_remote_object::<NodeInfo>(&well_known.links.href)?)
  54 }
  55
  56 /// Fetch any type of ActivityPub object, handling things like HTTP headers, deserialisation,
  57 /// timeouts etc.
  58 pub fn fetch_remote_object<Response>(url: &Url) -> Result<Response, Error>
  59 where
  60   Response: for<'de> Deserialize<'de>,
  61 {
  62   if !is_apub_id_valid(&url) {
  63     return Err(format_err!("Activitypub uri invalid or blocked: {}", url));
  64   }
  65   // TODO: this function should return a future
  66   let timeout = Duration::from_secs(60);
  67   let text = Request::get(url.as_str())
  68     .header("Accept", APUB_JSON_CONTENT_TYPE)
  69     .connect_timeout(timeout)
  70     .timeout(timeout)
  71     .body(())?
  72     .send()?
  73     .text()?;
  74   let res: Response = serde_json::from_str(&text)?;
  75   Ok(res)
  76 }
  77
  78 /// The types of ActivityPub objects that can be fetched directly by searching for their ID.
  79 #[serde(untagged)]
  80 #[derive(serde::Deserialize, Debug)]
  81 pub enum SearchAcceptedObjects {
  82   Person(Box<PersonExt>),
  83   Group(Box<GroupExt>),
  84   Page(Box<PageExt>),
  85   Comment(Box<Note>),
  86 }
  87
  88 /// Attempt to parse the query as URL, and fetch an ActivityPub object from it.
  89 ///
  90 /// Some working examples for use with the docker/federation/ setup:
  91 /// http://lemmy_alpha:8540/c/main, or !main@lemmy_alpha:8540
  92 /// http://lemmy_alpha:8540/u/lemmy_alpha, or @lemmy_alpha@lemmy_alpha:8540
  93 /// http://lemmy_alpha:8540/post/3
  94 /// http://lemmy_alpha:8540/comment/2
  95 pub fn search_by_apub_id(query: &str, conn: &PgConnection) -> Result<SearchResponse, Error> {
  96   // Parse the shorthand query url
  97   let query_url = if query.contains('@') {
  98     debug!("{}", query);
  99     let split = query.split('@').collect::<Vec<&str>>();
 100
 101     // User type will look like ['', username, instance]
 102     // Community will look like [!community, instance]
 103     let (name, instance) = if split.len() == 3 {
 104       (format!("/u/{}", split[1]), split[2])
 105     } else if split.len() == 2 {
 106       if split[0].contains('!') {
 107         let split2 = split[0].split('!').collect::<Vec<&str>>();
 108         (format!("/c/{}", split2[1]), split[1])
 109       } else {
 110         return Err(format_err!("Invalid search query: {}", query));
 111       }
 112     } else {
 113       return Err(format_err!("Invalid search query: {}", query));
 114     };
 115
 116     let url = format!("{}://{}{}", get_apub_protocol_string(), instance, name);
 117     Url::parse(&url)?
 118   } else {
 119     Url::parse(&query)?
 120   };
 121
 122   let mut response = SearchResponse {
 123     type_: SearchType::All.to_string(),
 124     comments: vec![],
 125     posts: vec![],
 126     communities: vec![],
 127     users: vec![],
 128   };
 129   match fetch_remote_object::<SearchAcceptedObjects>(&query_url)? {
 130     SearchAcceptedObjects::Person(p) => {
 131       let user_uri = p.inner.object_props.get_id().unwrap().to_string();
 132       let user = get_or_fetch_and_upsert_remote_user(&user_uri, &conn)?;
 133       response.users = vec![UserView::read(conn, user.id)?];
 134     }
 135     SearchAcceptedObjects::Group(g) => {
 136       let community_uri = g.inner.object_props.get_id().unwrap().to_string();
 137       let community = get_or_fetch_and_upsert_remote_community(&community_uri, &conn)?;
 138       // TODO Maybe at some point in the future, fetch all the history of a community
 139       // fetch_community_outbox(&c, conn)?;
 140       response.communities = vec![CommunityView::read(conn, community.id, None)?];
 141     }
 142     SearchAcceptedObjects::Page(p) => {
 143       let p = upsert_post(&PostForm::from_apub(&p, conn)?, conn)?;
 144       response.posts = vec![PostView::read(conn, p.id, None)?];
 145     }
 146     SearchAcceptedObjects::Comment(c) => {
 147       let post_url = c
 148         .object_props
 149         .get_many_in_reply_to_xsd_any_uris()
 150         .unwrap()
 151         .next()
 152         .unwrap()
 153         .to_string();
 154       // TODO: also fetch parent comments if any
 155       let post = fetch_remote_object(&Url::parse(&post_url)?)?;
 156       upsert_post(&PostForm::from_apub(&post, conn)?, conn)?;
 157       let c = upsert_comment(&CommentForm::from_apub(&c, conn)?, conn)?;
 158       response.comments = vec![CommentView::read(conn, c.id, None)?];
 159     }
 160   }
 161   Ok(response)
 162 }
 163
 164 /// Check if a remote user exists, create if not found, if its too old update it.Fetch a user, insert/update it in the database and return the user.
 165 pub fn get_or_fetch_and_upsert_remote_user(
 166   apub_id: &str,
 167   conn: &PgConnection,
 168 ) -> Result<User_, Error> {
 169   match User_::read_from_actor_id(&conn, &apub_id) {
 170     Ok(u) => {
 171       // If its older than a day, re-fetch it
 172       if !u.local && should_refetch_actor(u.last_refreshed_at) {
 173         debug!("Fetching and updating from remote user: {}", apub_id);
 174         let person = fetch_remote_object::<PersonExt>(&Url::parse(apub_id)?)?;
 175         let mut uf = UserForm::from_apub(&person, &conn)?;
 176         uf.last_refreshed_at = Some(naive_now());
 177         Ok(User_::update(&conn, u.id, &uf)?)
 178       } else {
 179         Ok(u)
 180       }
 181     }
 182     Err(NotFound {}) => {
 183       debug!("Fetching and creating remote user: {}", apub_id);
 184       let person = fetch_remote_object::<PersonExt>(&Url::parse(apub_id)?)?;
 185       let uf = UserForm::from_apub(&person, &conn)?;
 186       Ok(User_::create(conn, &uf)?)
 187     }
 188     Err(e) => Err(Error::from(e)),
 189   }
 190 }
 191
 192 /// Determines when a remote actor should be refetched from its instance. In release builds, this is
 193 /// ACTOR_REFETCH_INTERVAL_SECONDS after the last refetch, in debug builds always.
 194 ///
 195 /// TODO it won't pick up new avatars, summaries etc until a day after.
 196 /// Actors need an "update" activity pushed to other servers to fix this.
 197 fn should_refetch_actor(last_refreshed: NaiveDateTime) -> bool {
 198   if cfg!(debug_assertions) {
 199     true
 200   } else {
 201     let update_interval = chrono::Duration::seconds(ACTOR_REFETCH_INTERVAL_SECONDS);
 202     last_refreshed.lt(&(naive_now() - update_interval))
 203   }
 204 }
 205
 206 /// Check if a remote community exists, create if not found, if its too old update it.Fetch a community, insert/update it in the database and return the community.
 207 pub fn get_or_fetch_and_upsert_remote_community(
 208   apub_id: &str,
 209   conn: &PgConnection,
 210 ) -> Result<Community, Error> {
 211   match Community::read_from_actor_id(&conn, &apub_id) {
 212     Ok(c) => {
 213       if !c.local && should_refetch_actor(c.last_refreshed_at) {
 214         debug!("Fetching and updating from remote community: {}", apub_id);
 215         let group = fetch_remote_object::<GroupExt>(&Url::parse(apub_id)?)?;
 216         let mut cf = CommunityForm::from_apub(&group, conn)?;
 217         cf.last_refreshed_at = Some(naive_now());
 218         Ok(Community::update(&conn, c.id, &cf)?)
 219       } else {
 220         Ok(c)
 221       }
 222     }
 223     Err(NotFound {}) => {
 224       debug!("Fetching and creating remote community: {}", apub_id);
 225       let group = fetch_remote_object::<GroupExt>(&Url::parse(apub_id)?)?;
 226       let cf = CommunityForm::from_apub(&group, conn)?;
 227       let community = Community::create(conn, &cf)?;
 228
 229       // Also add the community moderators too
 230       let creator_and_moderator_uris = group
 231         .inner
 232         .object_props
 233         .get_many_attributed_to_xsd_any_uris()
 234         .unwrap();
 235       let creator_and_moderators = creator_and_moderator_uris
 236         .map(|c| get_or_fetch_and_upsert_remote_user(&c.to_string(), &conn).unwrap())
 237         .collect::<Vec<User_>>();
 238
 239       for mod_ in creator_and_moderators {
 240         let community_moderator_form = CommunityModeratorForm {
 241           community_id: community.id,
 242           user_id: mod_.id,
 243         };
 244         CommunityModerator::join(&conn, &community_moderator_form)?;
 245       }
 246
 247       Ok(community)
 248     }
 249     Err(e) => Err(Error::from(e)),
 250   }
 251 }
 252
 253 fn upsert_post(post_form: &PostForm, conn: &PgConnection) -> Result<Post, Error> {
 254   let existing = Post::read_from_apub_id(conn, &post_form.ap_id);
 255   match existing {
 256     Err(NotFound {}) => Ok(Post::create(conn, &post_form)?),
 257     Ok(p) => Ok(Post::update(conn, p.id, &post_form)?),
 258     Err(e) => Err(Error::from(e)),
 259   }
 260 }
 261
 262 pub fn get_or_fetch_and_insert_remote_post(
 263   post_ap_id: &str,
 264   conn: &PgConnection,
 265 ) -> Result<Post, Error> {
 266   match Post::read_from_apub_id(conn, post_ap_id) {
 267     Ok(p) => Ok(p),
 268     Err(NotFound {}) => {
 269       debug!("Fetching and creating remote post: {}", post_ap_id);
 270       let post = fetch_remote_object::<PageExt>(&Url::parse(post_ap_id)?)?;
 271       let post_form = PostForm::from_apub(&post, conn)?;
 272       Ok(Post::create(conn, &post_form)?)
 273     }
 274     Err(e) => Err(Error::from(e)),
 275   }
 276 }
 277
 278 fn upsert_comment(comment_form: &CommentForm, conn: &PgConnection) -> Result<Comment, Error> {
 279   let existing = Comment::read_from_apub_id(conn, &comment_form.ap_id);
 280   match existing {
 281     Err(NotFound {}) => Ok(Comment::create(conn, &comment_form)?),
 282     Ok(p) => Ok(Comment::update(conn, p.id, &comment_form)?),
 283     Err(e) => Err(Error::from(e)),
 284   }
 285 }
 286
 287 pub fn get_or_fetch_and_insert_remote_comment(
 288   comment_ap_id: &str,
 289   conn: &PgConnection,
 290 ) -> Result<Comment, Error> {
 291   match Comment::read_from_apub_id(conn, comment_ap_id) {
 292     Ok(p) => Ok(p),
 293     Err(NotFound {}) => {
 294       debug!(
 295         "Fetching and creating remote comment and its parents: {}",
 296         comment_ap_id
 297       );
 298       let comment = fetch_remote_object::<Note>(&Url::parse(comment_ap_id)?)?;
 299       let comment_form = CommentForm::from_apub(&comment, conn)?;
 300       Ok(Comment::create(conn, &comment_form)?)
 301     }
 302     Err(e) => Err(Error::from(e)),
 303   }
 304 }
 305
 306 // TODO It should not be fetching data from a community outbox.
 307 // All posts, comments, comment likes, etc should be posts to our community_inbox
 308 // The only data we should be periodically fetching (if it hasn't been fetched in the last day
 309 // maybe), is community and user actors
 310 // and user actors
 311 // Fetch all posts in the outbox of the given user, and insert them into the database.
 312 // fn fetch_community_outbox(community: &Community, conn: &PgConnection) -> Result<Vec<Post>, Error> {
 313 //   let outbox_url = Url::parse(&community.get_outbox_url())?;
 314 //   let outbox = fetch_remote_object::<OrderedCollection>(&outbox_url)?;
 315 //   let items = outbox.collection_props.get_many_items_base_boxes();
 316
 317 //   Ok(
 318 //     items
 319 //       .unwrap()
 320 //       .map(|obox: &BaseBox| -> Result<PostForm, Error> {
 321 //         let page = obox.clone().to_concrete::<Page>()?;
 322 //         PostForm::from_page(&page, conn)
 323 //       })
 324 //       .map(|pf| upsert_post(&pf?, conn))
 325 //       .collect::<Result<Vec<Post>, Error>>()?,
 326 //   )
 327 // }