2 check_is_apub_id_valid,
9 APUB_JSON_CONTENT_TYPE,
11 use activitystreams::{base::BaseExt, collection::OrderedCollection, prelude::*};
12 use anyhow::{anyhow, Context};
13 use chrono::NaiveDateTime;
14 use diesel::result::Error::NotFound;
17 comment_view::CommentView,
18 community::{Community, CommunityModerator, CommunityModeratorForm},
19 community_view::CommunityView,
29 use lemmy_structs::{blocking, site::SearchResponse};
32 request::{retry, RecvError},
36 use lemmy_websocket::LemmyContext;
39 use serde::Deserialize;
40 use std::{fmt::Debug, time::Duration};
43 static ACTOR_REFETCH_INTERVAL_SECONDS: i64 = 24 * 60 * 60;
44 static ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG: i64 = 10;
46 /// Maximum number of HTTP requests allowed to handle a single incoming activity (or a single object
47 /// fetch through the search).
49 /// Tests are passing with a value of 5, so 10 should be safe for production.
50 static MAX_REQUEST_NUMBER: i32 = 10;
52 /// Fetch any type of ActivityPub object, handling things like HTTP headers, deserialisation,
54 async fn fetch_remote_object<Response>(
57 recursion_counter: &mut i32,
58 ) -> Result<Response, LemmyError>
60 Response: for<'de> Deserialize<'de>,
62 *recursion_counter += 1;
63 if *recursion_counter > MAX_REQUEST_NUMBER {
64 return Err(anyhow!("Maximum recursion depth reached").into());
66 check_is_apub_id_valid(&url)?;
68 let timeout = Duration::from_secs(60);
73 .header("Accept", APUB_JSON_CONTENT_TYPE)
81 debug!("Receive error, {}", e);
82 RecvError(e.to_string())
88 /// The types of ActivityPub objects that can be fetched directly by searching for their ID.
90 #[derive(serde::Deserialize, Debug)]
91 enum SearchAcceptedObjects {
92 Person(Box<PersonExt>),
95 Comment(Box<NoteExt>),
98 /// Attempt to parse the query as URL, and fetch an ActivityPub object from it.
100 /// Some working examples for use with the `docker/federation/` setup:
101 /// http://lemmy_alpha:8541/c/main, or !main@lemmy_alpha:8541
102 /// http://lemmy_beta:8551/u/lemmy_alpha, or @lemmy_beta@lemmy_beta:8551
103 /// http://lemmy_gamma:8561/post/3
104 /// http://lemmy_delta:8571/comment/2
105 pub async fn search_by_apub_id(
107 context: &LemmyContext,
108 ) -> Result<SearchResponse, LemmyError> {
109 // Parse the shorthand query url
110 let query_url = if query.contains('@') {
111 debug!("Search for {}", query);
112 let split = query.split('@').collect::<Vec<&str>>();
114 // User type will look like ['', username, instance]
115 // Community will look like [!community, instance]
116 let (name, instance) = if split.len() == 3 {
117 (format!("/u/{}", split[1]), split[2])
118 } else if split.len() == 2 {
119 if split[0].contains('!') {
120 let split2 = split[0].split('!').collect::<Vec<&str>>();
121 (format!("/c/{}", split2[1]), split[1])
123 return Err(anyhow!("Invalid search query: {}", query).into());
126 return Err(anyhow!("Invalid search query: {}", query).into());
131 Settings::get().get_protocol_string(),
140 let mut response = SearchResponse {
141 type_: SearchType::All.to_string(),
148 let domain = query_url.domain().context("url has no domain")?;
149 let recursion_counter = &mut 0;
150 let response = match fetch_remote_object::<SearchAcceptedObjects>(
157 SearchAcceptedObjects::Person(p) => {
158 let user_uri = p.inner.id(domain)?.context("person has no id")?;
160 let user = get_or_fetch_and_upsert_user(&user_uri, context, recursion_counter).await?;
162 response.users = vec![
163 blocking(context.pool(), move |conn| {
164 UserView::get_user_secure(conn, user.id)
171 SearchAcceptedObjects::Group(g) => {
172 let community_uri = g.inner.id(domain)?.context("group has no id")?;
175 get_or_fetch_and_upsert_community(community_uri, context, recursion_counter).await?;
177 response.communities = vec![
178 blocking(context.pool(), move |conn| {
179 CommunityView::read(conn, community.id, None)
186 SearchAcceptedObjects::Page(p) => {
187 let p = Post::from_apub(&p, context, query_url, recursion_counter).await?;
190 vec![blocking(context.pool(), move |conn| PostView::read(conn, p.id, None)).await??];
194 SearchAcceptedObjects::Comment(c) => {
195 let c = Comment::from_apub(&c, context, query_url, recursion_counter).await?;
197 response.comments = vec![
198 blocking(context.pool(), move |conn| {
199 CommentView::read(conn, c.id, None)
211 /// Get a remote actor from its apub ID (either a user or a community). Thin wrapper around
212 /// `get_or_fetch_and_upsert_user()` and `get_or_fetch_and_upsert_community()`.
214 /// If it exists locally and `!should_refetch_actor()`, it is returned directly from the database.
215 /// Otherwise it is fetched from the remote instance, stored and returned.
216 pub(crate) async fn get_or_fetch_and_upsert_actor(
218 context: &LemmyContext,
219 recursion_counter: &mut i32,
220 ) -> Result<Box<dyn ActorType>, LemmyError> {
221 let community = get_or_fetch_and_upsert_community(apub_id, context, recursion_counter).await;
222 let actor: Box<dyn ActorType> = match community {
223 Ok(c) => Box::new(c),
224 Err(_) => Box::new(get_or_fetch_and_upsert_user(apub_id, context, recursion_counter).await?),
229 /// Get a user from its apub ID.
231 /// If it exists locally and `!should_refetch_actor()`, it is returned directly from the database.
232 /// Otherwise it is fetched from the remote instance, stored and returned.
233 pub(crate) async fn get_or_fetch_and_upsert_user(
235 context: &LemmyContext,
236 recursion_counter: &mut i32,
237 ) -> Result<User_, LemmyError> {
238 let apub_id_owned = apub_id.to_owned();
239 let user = blocking(context.pool(), move |conn| {
240 User_::read_from_apub_id(conn, apub_id_owned.as_ref())
245 // If its older than a day, re-fetch it
246 Ok(u) if !u.local && should_refetch_actor(u.last_refreshed_at) => {
247 debug!("Fetching and updating from remote user: {}", apub_id);
249 fetch_remote_object::<PersonExt>(context.client(), apub_id, recursion_counter).await;
250 // If fetching failed, return the existing data.
255 let user = User_::from_apub(&person?, context, apub_id.to_owned(), recursion_counter).await?;
257 let user_id = user.id;
258 blocking(context.pool(), move |conn| {
259 User_::mark_as_updated(conn, user_id)
266 Err(NotFound {}) => {
267 debug!("Fetching and creating remote user: {}", apub_id);
269 fetch_remote_object::<PersonExt>(context.client(), apub_id, recursion_counter).await?;
271 let user = User_::from_apub(&person, context, apub_id.to_owned(), recursion_counter).await?;
275 Err(e) => Err(e.into()),
279 /// Determines when a remote actor should be refetched from its instance. In release builds, this is
280 /// `ACTOR_REFETCH_INTERVAL_SECONDS` after the last refetch, in debug builds
281 /// `ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG`.
283 /// TODO it won't pick up new avatars, summaries etc until a day after.
284 /// Actors need an "update" activity pushed to other servers to fix this.
285 fn should_refetch_actor(last_refreshed: NaiveDateTime) -> bool {
286 let update_interval = if cfg!(debug_assertions) {
287 // avoid infinite loop when fetching community outbox
288 chrono::Duration::seconds(ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG)
290 chrono::Duration::seconds(ACTOR_REFETCH_INTERVAL_SECONDS)
292 last_refreshed.lt(&(naive_now() - update_interval))
295 /// Get a community from its apub ID.
297 /// If it exists locally and `!should_refetch_actor()`, it is returned directly from the database.
298 /// Otherwise it is fetched from the remote instance, stored and returned.
299 pub(crate) async fn get_or_fetch_and_upsert_community(
301 context: &LemmyContext,
302 recursion_counter: &mut i32,
303 ) -> Result<Community, LemmyError> {
304 let apub_id_owned = apub_id.to_owned();
305 let community = blocking(context.pool(), move |conn| {
306 Community::read_from_apub_id(conn, apub_id_owned.as_str())
311 Ok(c) if !c.local && should_refetch_actor(c.last_refreshed_at) => {
312 debug!("Fetching and updating from remote community: {}", apub_id);
313 fetch_remote_community(apub_id, context, Some(c), recursion_counter).await
316 Err(NotFound {}) => {
317 debug!("Fetching and creating remote community: {}", apub_id);
318 fetch_remote_community(apub_id, context, None, recursion_counter).await
320 Err(e) => Err(e.into()),
324 /// Request a community by apub ID from a remote instance, including moderators. If `old_community`,
325 /// is set, this is an update for a community which is already known locally. If not, we don't know
326 /// the community yet and also pull the outbox, to get some initial posts.
327 async fn fetch_remote_community(
329 context: &LemmyContext,
330 old_community: Option<Community>,
331 recursion_counter: &mut i32,
332 ) -> Result<Community, LemmyError> {
333 let group = fetch_remote_object::<GroupExt>(context.client(), apub_id, recursion_counter).await;
334 // If fetching failed, return the existing data.
335 if let Some(ref c) = old_community {
337 return Ok(c.to_owned());
343 Community::from_apub(&group, context, apub_id.to_owned(), recursion_counter).await?;
345 // Also add the community moderators too
346 let attributed_to = group.inner.attributed_to().context(location_info!())?;
347 let creator_and_moderator_uris: Vec<&Url> = attributed_to
349 .context(location_info!())?
351 .map(|a| a.as_xsd_any_uri().context(""))
352 .collect::<Result<Vec<&Url>, anyhow::Error>>()?;
354 let mut creator_and_moderators = Vec::new();
356 for uri in creator_and_moderator_uris {
357 let c_or_m = get_or_fetch_and_upsert_user(uri, context, recursion_counter).await?;
359 creator_and_moderators.push(c_or_m);
362 // TODO: need to make this work to update mods of existing communities
363 if old_community.is_none() {
364 let community_id = community.id;
365 blocking(context.pool(), move |conn| {
366 for mod_ in creator_and_moderators {
367 let community_moderator_form = CommunityModeratorForm {
372 CommunityModerator::join(conn, &community_moderator_form)?;
374 Ok(()) as Result<(), LemmyError>
379 // fetch outbox (maybe make this conditional)
380 let outbox = fetch_remote_object::<OrderedCollection>(
382 &community.get_outbox_url()?,
386 let outbox_items = outbox.items().context(location_info!())?.clone();
387 let mut outbox_items = outbox_items.many().context(location_info!())?;
388 if outbox_items.len() > 20 {
389 outbox_items = outbox_items[0..20].to_vec();
391 for o in outbox_items {
392 let page = PageExt::from_any_base(o)?.context(location_info!())?;
393 let page_id = page.id_unchecked().context(location_info!())?;
395 // The post creator may be from a blocked instance, if it errors, then skip it
396 if check_is_apub_id_valid(page_id).is_err() {
399 Post::from_apub(&page, context, page_id.to_owned(), recursion_counter).await?;
400 // TODO: we need to send a websocket update here
406 /// Gets a post by its apub ID. If it exists locally, it is returned directly. Otherwise it is
407 /// pulled from its apub ID, inserted and returned.
409 /// The parent community is also pulled if necessary. Comments are not pulled.
410 pub(crate) async fn get_or_fetch_and_insert_post(
412 context: &LemmyContext,
413 recursion_counter: &mut i32,
414 ) -> Result<Post, LemmyError> {
415 let post_ap_id_owned = post_ap_id.to_owned();
416 let post = blocking(context.pool(), move |conn| {
417 Post::read_from_apub_id(conn, post_ap_id_owned.as_str())
423 Err(NotFound {}) => {
424 debug!("Fetching and creating remote post: {}", post_ap_id);
426 fetch_remote_object::<PageExt>(context.client(), post_ap_id, recursion_counter).await?;
427 let post = Post::from_apub(&page, context, post_ap_id.to_owned(), recursion_counter).await?;
431 Err(e) => Err(e.into()),
435 /// Gets a comment by its apub ID. If it exists locally, it is returned directly. Otherwise it is
436 /// pulled from its apub ID, inserted and returned.
438 /// The parent community, post and comment are also pulled if necessary.
439 pub(crate) async fn get_or_fetch_and_insert_comment(
441 context: &LemmyContext,
442 recursion_counter: &mut i32,
443 ) -> Result<Comment, LemmyError> {
444 let comment_ap_id_owned = comment_ap_id.to_owned();
445 let comment = blocking(context.pool(), move |conn| {
446 Comment::read_from_apub_id(conn, comment_ap_id_owned.as_str())
452 Err(NotFound {}) => {
454 "Fetching and creating remote comment and its parents: {}",
458 fetch_remote_object::<NoteExt>(context.client(), comment_ap_id, recursion_counter).await?;
459 let comment = Comment::from_apub(
462 comment_ap_id.to_owned(),
467 let post_id = comment.post_id;
468 let post = blocking(context.pool(), move |conn| Post::read(conn, post_id)).await??;
470 return Err(anyhow!("Post is locked").into());
475 Err(e) => Err(e.into()),