mirror of https://github.com/LemmyNet/lemmy.git
rewrite community to use new fetcher
parent
27e409442a
commit
1013d1a46d
|
@ -9,7 +9,7 @@ use crate::{
|
|||
},
|
||||
activity_queue::send_to_community_new,
|
||||
extensions::context::lemmy_context,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
generate_moderators_url,
|
||||
ActorType,
|
||||
};
|
||||
|
@ -93,8 +93,7 @@ impl ActivityHandler for AddMod {
|
|||
context: &LemmyContext,
|
||||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
let community =
|
||||
get_or_fetch_and_upsert_community(&self.cc[0], context, request_counter).await?;
|
||||
let community = dereference::<Community>(&self.cc[0], context, request_counter).await?;
|
||||
let new_mod = dereference::<Person>(&self.object, context, request_counter).await?;
|
||||
|
||||
// If we had to refetch the community while parsing the activity, then the new mod has already
|
||||
|
|
|
@ -8,7 +8,7 @@ use crate::{
|
|||
},
|
||||
activity_queue::send_to_community_new,
|
||||
extensions::context::lemmy_context,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
ActorType,
|
||||
};
|
||||
use activitystreams::{
|
||||
|
@ -102,8 +102,7 @@ impl ActivityHandler for BlockUserFromCommunity {
|
|||
context: &LemmyContext,
|
||||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
let community =
|
||||
get_or_fetch_and_upsert_community(&self.cc[0], context, request_counter).await?;
|
||||
let community = dereference::<Community>(&self.cc[0], context, request_counter).await?;
|
||||
let blocked_user = dereference::<Person>(&self.object, context, request_counter).await?;
|
||||
|
||||
let community_user_ban_form = CommunityPersonBanForm {
|
||||
|
|
|
@ -10,7 +10,7 @@ use crate::{
|
|||
},
|
||||
activity_queue::send_to_community_new,
|
||||
extensions::context::lemmy_context,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
generate_moderators_url,
|
||||
ActorType,
|
||||
};
|
||||
|
@ -108,8 +108,7 @@ impl ActivityHandler for RemoveMod {
|
|||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
if self.target.is_some() {
|
||||
let community =
|
||||
get_or_fetch_and_upsert_community(&self.cc[0], context, request_counter).await?;
|
||||
let community = dereference::<Community>(&self.cc[0], context, request_counter).await?;
|
||||
let remove_mod = dereference::<Person>(&self.object, context, request_counter).await?;
|
||||
|
||||
let form = CommunityModeratorForm {
|
||||
|
|
|
@ -8,7 +8,7 @@ use crate::{
|
|||
},
|
||||
activity_queue::send_to_community_new,
|
||||
extensions::context::lemmy_context,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
ActorType,
|
||||
};
|
||||
use activitystreams::{
|
||||
|
@ -91,8 +91,7 @@ impl ActivityHandler for UndoBlockUserFromCommunity {
|
|||
context: &LemmyContext,
|
||||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
let community =
|
||||
get_or_fetch_and_upsert_community(&self.cc[0], context, request_counter).await?;
|
||||
let community = dereference::<Community>(&self.cc[0], context, request_counter).await?;
|
||||
let blocked_user = dereference::<Person>(&self.object.object, context, request_counter).await?;
|
||||
|
||||
let community_user_ban_form = CommunityPersonBanForm {
|
||||
|
|
|
@ -7,7 +7,7 @@ use crate::{
|
|||
},
|
||||
activity_queue::send_activity_new,
|
||||
extensions::context::lemmy_context,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
ActorType,
|
||||
};
|
||||
use activitystreams::{
|
||||
|
@ -90,7 +90,7 @@ impl ActivityHandler for AcceptFollowCommunity {
|
|||
context: &LemmyContext,
|
||||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
let actor = get_or_fetch_and_upsert_community(&self.actor, context, request_counter).await?;
|
||||
let actor = dereference::<Community>(&self.actor, context, request_counter).await?;
|
||||
let to = dereference::<Person>(&self.to, context, request_counter).await?;
|
||||
// This will throw an error if no follow was requested
|
||||
blocking(context.pool(), move |conn| {
|
||||
|
|
|
@ -7,7 +7,7 @@ use crate::{
|
|||
},
|
||||
activity_queue::send_activity_new,
|
||||
extensions::context::lemmy_context,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
ActorType,
|
||||
};
|
||||
use activitystreams::{
|
||||
|
@ -98,8 +98,7 @@ impl ActivityHandler for FollowCommunity {
|
|||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
let actor = dereference::<Person>(&self.actor, context, request_counter).await?;
|
||||
let community =
|
||||
get_or_fetch_and_upsert_community(&self.object, context, request_counter).await?;
|
||||
let community = dereference::<Community>(&self.object, context, request_counter).await?;
|
||||
let community_follower_form = CommunityFollowerForm {
|
||||
community_id: community.id,
|
||||
person_id: actor.id,
|
||||
|
|
|
@ -7,7 +7,7 @@ use crate::{
|
|||
},
|
||||
activity_queue::send_activity_new,
|
||||
extensions::context::lemmy_context,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
ActorType,
|
||||
};
|
||||
use activitystreams::{
|
||||
|
@ -85,7 +85,7 @@ impl ActivityHandler for UndoFollowCommunity {
|
|||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
let actor = dereference::<Person>(&self.actor, context, request_counter).await?;
|
||||
let community = get_or_fetch_and_upsert_community(&self.to, context, request_counter).await?;
|
||||
let community = dereference::<Community>(&self.to, context, request_counter).await?;
|
||||
|
||||
let community_follower_form = CommunityFollowerForm {
|
||||
community_id: community.id,
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
use crate::{
|
||||
check_community_or_site_ban,
|
||||
check_is_apub_id_valid,
|
||||
fetcher::{community::get_or_fetch_and_upsert_community, new_fetcher::dereference},
|
||||
fetcher::new_fetcher::dereference,
|
||||
generate_moderators_url,
|
||||
};
|
||||
use anyhow::anyhow;
|
||||
|
@ -58,7 +58,7 @@ pub(crate) async fn extract_community(
|
|||
let mut cc_iter = cc.iter();
|
||||
loop {
|
||||
if let Some(cid) = cc_iter.next() {
|
||||
if let Ok(c) = get_or_fetch_and_upsert_community(cid, context, request_counter).await {
|
||||
if let Ok(c) = dereference(cid, context, request_counter).await {
|
||||
break Ok(c);
|
||||
}
|
||||
} else {
|
||||
|
@ -75,7 +75,7 @@ pub(crate) async fn verify_person_in_community(
|
|||
context: &LemmyContext,
|
||||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
let community = get_or_fetch_and_upsert_community(community_id, context, request_counter).await?;
|
||||
let community = dereference::<Community>(community_id, context, request_counter).await?;
|
||||
let person = dereference::<Person>(person_id, context, request_counter).await?;
|
||||
check_community_or_site_ban(&person, community.id, context.pool()).await
|
||||
}
|
||||
|
@ -86,7 +86,7 @@ async fn verify_community(
|
|||
context: &LemmyContext,
|
||||
request_counter: &mut i32,
|
||||
) -> Result<(), LemmyError> {
|
||||
get_or_fetch_and_upsert_community(community_id, context, request_counter).await?;
|
||||
dereference::<Community>(community_id, context, request_counter).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
|
@ -1,19 +1,13 @@
|
|||
use crate::{
|
||||
activities::community::announce::AnnounceActivity,
|
||||
fetcher::{
|
||||
fetch::fetch_remote_object,
|
||||
is_deleted,
|
||||
new_fetcher::dereference,
|
||||
should_refetch_actor,
|
||||
},
|
||||
objects::{community::Group, FromApub},
|
||||
fetcher::{fetch::fetch_remote_object, new_fetcher::dereference},
|
||||
objects::community::Group,
|
||||
};
|
||||
use activitystreams::collection::{CollectionExt, OrderedCollection};
|
||||
use anyhow::Context;
|
||||
use diesel::result::Error::NotFound;
|
||||
use lemmy_api_common::blocking;
|
||||
use lemmy_apub_lib::ActivityHandler;
|
||||
use lemmy_db_queries::{source::community::Community_, ApubObject, Joinable};
|
||||
use lemmy_db_queries::Joinable;
|
||||
use lemmy_db_schema::source::{
|
||||
community::{Community, CommunityModerator, CommunityModeratorForm},
|
||||
person::Person,
|
||||
|
@ -21,75 +15,9 @@ use lemmy_db_schema::source::{
|
|||
use lemmy_db_views_actor::community_moderator_view::CommunityModeratorView;
|
||||
use lemmy_utils::{location_info, LemmyError};
|
||||
use lemmy_websocket::LemmyContext;
|
||||
use log::debug;
|
||||
use url::Url;
|
||||
|
||||
/// Get a community from its apub ID.
|
||||
///
|
||||
/// If it exists locally and `!should_refetch_actor()`, it is returned directly from the database.
|
||||
/// Otherwise it is fetched from the remote instance, stored and returned.
|
||||
pub(crate) async fn get_or_fetch_and_upsert_community(
|
||||
apub_id: &Url,
|
||||
context: &LemmyContext,
|
||||
recursion_counter: &mut i32,
|
||||
) -> Result<Community, LemmyError> {
|
||||
let apub_id_owned = apub_id.to_owned();
|
||||
let community = blocking(context.pool(), move |conn| {
|
||||
Community::read_from_apub_id(conn, &apub_id_owned.into())
|
||||
})
|
||||
.await?;
|
||||
|
||||
match community {
|
||||
Ok(c) if !c.local && should_refetch_actor(c.last_refreshed_at) => {
|
||||
debug!("Fetching and updating from remote community: {}", apub_id);
|
||||
fetch_remote_community(apub_id, context, Some(c), recursion_counter).await
|
||||
}
|
||||
Ok(c) => Ok(c),
|
||||
Err(NotFound {}) => {
|
||||
debug!("Fetching and creating remote community: {}", apub_id);
|
||||
fetch_remote_community(apub_id, context, None, recursion_counter).await
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Request a community by apub ID from a remote instance, including moderators. If `old_community`,
|
||||
/// is set, this is an update for a community which is already known locally. If not, we don't know
|
||||
/// the community yet and also pull the outbox, to get some initial posts.
|
||||
async fn fetch_remote_community(
|
||||
apub_id: &Url,
|
||||
context: &LemmyContext,
|
||||
old_community: Option<Community>,
|
||||
request_counter: &mut i32,
|
||||
) -> Result<Community, LemmyError> {
|
||||
let group = fetch_remote_object::<Group>(context.client(), apub_id, request_counter).await;
|
||||
|
||||
if let Some(c) = old_community.to_owned() {
|
||||
if is_deleted(&group) {
|
||||
blocking(context.pool(), move |conn| {
|
||||
Community::update_deleted(conn, c.id, true)
|
||||
})
|
||||
.await??;
|
||||
} else if group.is_err() {
|
||||
// If fetching failed, return the existing data.
|
||||
return Ok(c);
|
||||
}
|
||||
}
|
||||
|
||||
let group = group?;
|
||||
let community = Community::from_apub(&group, context, apub_id, request_counter).await?;
|
||||
|
||||
update_community_mods(&group, &community, context, request_counter).await?;
|
||||
|
||||
// only fetch outbox for new communities, otherwise this can create an infinite loop
|
||||
if old_community.is_none() {
|
||||
fetch_community_outbox(context, &group.outbox, request_counter).await?
|
||||
}
|
||||
|
||||
Ok(community)
|
||||
}
|
||||
|
||||
async fn update_community_mods(
|
||||
pub(crate) async fn update_community_mods(
|
||||
group: &Group,
|
||||
community: &Community,
|
||||
context: &LemmyContext,
|
||||
|
@ -139,7 +67,7 @@ async fn update_community_mods(
|
|||
Ok(())
|
||||
}
|
||||
|
||||
async fn fetch_community_outbox(
|
||||
pub(crate) async fn fetch_community_outbox(
|
||||
context: &LemmyContext,
|
||||
outbox: &Url,
|
||||
recursion_counter: &mut i32,
|
||||
|
|
|
@ -5,16 +5,15 @@ pub mod post_or_comment;
|
|||
pub mod search;
|
||||
|
||||
use crate::{
|
||||
fetcher::{
|
||||
community::get_or_fetch_and_upsert_community,
|
||||
fetch::FetchError,
|
||||
new_fetcher::dereference,
|
||||
},
|
||||
fetcher::{fetch::FetchError, new_fetcher::dereference},
|
||||
ActorType,
|
||||
};
|
||||
use chrono::NaiveDateTime;
|
||||
use http::StatusCode;
|
||||
use lemmy_db_schema::{naive_now, source::person::Person};
|
||||
use lemmy_db_schema::{
|
||||
naive_now,
|
||||
source::{community::Community, person::Person},
|
||||
};
|
||||
use lemmy_utils::LemmyError;
|
||||
use lemmy_websocket::LemmyContext;
|
||||
use serde::Deserialize;
|
||||
|
@ -47,7 +46,7 @@ pub(crate) async fn get_or_fetch_and_upsert_actor(
|
|||
context: &LemmyContext,
|
||||
recursion_counter: &mut i32,
|
||||
) -> Result<Box<dyn ActorType>, LemmyError> {
|
||||
let community = get_or_fetch_and_upsert_community(apub_id, context, recursion_counter).await;
|
||||
let community = dereference::<Community>(apub_id, context, recursion_counter).await;
|
||||
let actor: Box<dyn ActorType> = match community {
|
||||
Ok(c) => Box::new(c),
|
||||
Err(_) => Box::new(dereference::<Person>(apub_id, context, recursion_counter).await?),
|
||||
|
|
|
@ -27,7 +27,6 @@ where
|
|||
let db_object = dereference_locally::<Kind>(id.clone(), context.pool()).await?;
|
||||
// if its a local object, only fetch it from the database and not over http
|
||||
if id.domain() == Some(&Settings::get().get_hostname_without_port()?) {
|
||||
dbg!("is local object", db_object.is_some());
|
||||
return match db_object {
|
||||
None => Err(NotFound {}.into()),
|
||||
Some(o) => Ok(o),
|
||||
|
|
|
@ -1,10 +1,5 @@
|
|||
use crate::{
|
||||
fetcher::{
|
||||
community::get_or_fetch_and_upsert_community,
|
||||
fetch::fetch_remote_object,
|
||||
is_deleted,
|
||||
new_fetcher::dereference,
|
||||
},
|
||||
fetcher::{fetch::fetch_remote_object, is_deleted, new_fetcher::dereference},
|
||||
find_object_by_id,
|
||||
objects::{comment::Note, community::Group, person::Person as ApubPerson, post::Page, FromApub},
|
||||
Object,
|
||||
|
@ -143,8 +138,7 @@ async fn build_response(
|
|||
}
|
||||
SearchAcceptedObjects::Group(g) => {
|
||||
let community_uri = g.id(&query_url)?;
|
||||
let community =
|
||||
get_or_fetch_and_upsert_community(community_uri, context, recursion_counter).await?;
|
||||
let community = dereference::<Community>(community_uri, context, recursion_counter).await?;
|
||||
ROR {
|
||||
community: blocking(context.pool(), move |conn| {
|
||||
CommunityView::read(conn, community.id, None)
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
use crate::{
|
||||
extensions::{context::lemmy_context, signatures::PublicKey},
|
||||
fetcher::community::fetch_community_mods,
|
||||
fetcher::community::{fetch_community_mods, fetch_community_outbox, update_community_mods},
|
||||
generate_moderators_url,
|
||||
objects::{create_tombstone, FromApub, ImageObject, Source, ToApub},
|
||||
ActorType,
|
||||
|
@ -179,6 +179,11 @@ impl FromApub for Community {
|
|||
let form = Group::from_apub_to_form(group, expected_domain).await?;
|
||||
|
||||
let community = blocking(context.pool(), move |conn| Community::upsert(conn, &form)).await??;
|
||||
update_community_mods(group, &community, context, request_counter).await?;
|
||||
|
||||
// TODO: doing this unconditionally might cause infinite loop for some reason
|
||||
fetch_community_outbox(context, &group.outbox, request_counter).await?;
|
||||
|
||||
Ok(community)
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue