lemmy/crates/db_views_actor/src/community_follower_view.rs
phiresky 375d9a2a3c
Persistent, performant, reliable federation queue (#3605)
* persistent activity queue

* fixes

* fixes

* make federation workers function callable from outside

* log federation instances

* dead instance detection not needed here

* taplo fmt

* split federate bin/lib

* minor fix

* better logging

* log

* create struct to hold cancellable task for readability

* use boxfuture for readability

* reset submodule

* fix

* fix lint

* swap

* remove json column, use separate array columns instead

* some review comments

* make worker a struct for readability

* minor readability

* add local filter to community follower view

* remove separate lemmy_federate entry point

* fix remaining duration

* address review comments mostly

* fix lint

* upgrade actitypub-fed to simpler interface

* fix sql format

* increase delays a bit

* fixes after merge

* remove selectable

* fix instance selectable

* add comment

* start federation based on latest id at the time

* rename federate process args

* dead instances in one query

* filter follow+report activities by local

* remove synchronous federation

remove activity sender queue

* lint

* fix federation tests by waiting for results to change

* fix fed test

* fix comment report

* wait some more

* Apply suggestions from code review

Co-authored-by: SorteKanin <sortekanin@gmail.com>

* fix most remaining tests

* wait until private messages

* fix community tests

* fix community tests

* move arg parse

* use instance_id instead of domain in federation_queue_state table

---------

Co-authored-by: Dessalines <dessalines@users.noreply.github.com>
Co-authored-by: SorteKanin <sortekanin@gmail.com>
2023-09-09 12:25:03 -04:00

88 lines
3.3 KiB
Rust

use crate::structs::CommunityFollowerView;
use chrono::Utc;
use diesel::{
dsl::{count_star, not},
result::Error,
ExpressionMethods,
QueryDsl,
};
use diesel_async::RunQueryDsl;
use lemmy_db_schema::{
newtypes::{CommunityId, DbUrl, InstanceId, PersonId},
schema::{community, community_follower, person},
utils::{functions::coalesce, get_conn, DbPool},
};
impl CommunityFollowerView {
/// return a list of local community ids and remote inboxes that at least one user of the given instance has followed
pub async fn get_instance_followed_community_inboxes(
pool: &mut DbPool<'_>,
instance_id: InstanceId,
published_since: chrono::DateTime<Utc>,
) -> Result<Vec<(CommunityId, DbUrl)>, Error> {
let conn = &mut get_conn(pool).await?;
// In most cases this will fetch the same url many times (the shared inbox url)
// PG will only send a single copy to rust, but it has to scan through all follower rows (same as it was before).
// So on the PG side it would be possible to optimize this further by adding e.g. a new table community_followed_instances (community_id, instance_id)
// that would work for all instances that support fully shared inboxes.
// It would be a bit more complicated though to keep it in sync.
community_follower::table
.inner_join(community::table)
.inner_join(person::table)
.filter(person::instance_id.eq(instance_id))
.filter(community::local) // this should be a no-op since community_followers table only has local-person+remote-community or remote-person+local-community
.filter(not(person::local))
.filter(community_follower::published.gt(published_since.naive_utc()))
.select((
community::id,
coalesce(person::shared_inbox_url, person::inbox_url),
))
.distinct() // only need each community_id, inbox combination once
.load::<(CommunityId, DbUrl)>(conn)
.await
}
pub async fn get_community_follower_inboxes(
pool: &mut DbPool<'_>,
community_id: CommunityId,
) -> Result<Vec<DbUrl>, Error> {
let conn = &mut get_conn(pool).await?;
let res = community_follower::table
.filter(community_follower::community_id.eq(community_id))
.filter(not(person::local))
.inner_join(person::table)
.select(coalesce(person::shared_inbox_url, person::inbox_url))
.distinct()
.load::<DbUrl>(conn)
.await?;
Ok(res)
}
pub async fn count_community_followers(
pool: &mut DbPool<'_>,
community_id: CommunityId,
) -> Result<i64, Error> {
let conn = &mut get_conn(pool).await?;
let res = community_follower::table
.filter(community_follower::community_id.eq(community_id))
.select(count_star())
.first::<i64>(conn)
.await?;
Ok(res)
}
pub async fn for_person(pool: &mut DbPool<'_>, person_id: PersonId) -> Result<Vec<Self>, Error> {
let conn = &mut get_conn(pool).await?;
community_follower::table
.inner_join(community::table)
.inner_join(person::table)
.select((community::all_columns, person::all_columns))
.filter(community_follower::person_id.eq(person_id))
.filter(community::deleted.eq(false))
.filter(community::removed.eq(false))
.order_by(community::title)
.load::<CommunityFollowerView>(conn)
.await
}
}