crates/db_schema/src/impls/instance.rs

   1 use crate::{
   2   newtypes::InstanceId,
   3   schema::{federation_allowlist, federation_blocklist, instance},
   4   source::instance::{Instance, InstanceForm},
   5   utils::{get_conn, naive_now, DbPool},
   6 };
   7 use diesel::{dsl::insert_into, result::Error, ExpressionMethods, QueryDsl};
   8 use diesel_async::{AsyncPgConnection, RunQueryDsl};
   9
  10 impl Instance {
  11   pub(crate) async fn read_or_create_with_conn(
  12     conn: &mut AsyncPgConnection,
  13     domain_: String,
  14   ) -> Result<Self, Error> {
  15     use crate::schema::instance::domain;
  16     // First try to read the instance row and return directly if found
  17     let instance = instance::table
  18       .filter(domain.eq(&domain_))
  19       .first::<Self>(conn)
  20       .await;
  21     match instance {
  22       Ok(i) => Ok(i),
  23       Err(diesel::NotFound) => {
  24         // Instance not in database yet, insert it
  25         let form = InstanceForm::builder()
  26           .domain(domain_)
  27           .updated(Some(naive_now()))
  28           .build();
  29         insert_into(instance::table)
  30           .values(&form)
  31           // Necessary because this method may be called concurrently for the same domain. This
  32           // could be handled with a transaction, but nested transactions arent allowed
  33           .on_conflict(instance::domain)
  34           .do_update()
  35           .set(&form)
  36           .get_result::<Self>(conn)
  37           .await
  38       }
  39       e => e,
  40     }
  41   }
  42
  43   /// Attempt to read Instance column for the given domain. If it doesnt exist, insert a new one.
  44   /// There is no need for update as the domain of an existing instance cant change.
  45   pub async fn read_or_create(pool: &DbPool, domain: String) -> Result<Self, Error> {
  46     let conn = &mut get_conn(pool).await?;
  47     Self::read_or_create_with_conn(conn, domain).await
  48   }
  49   pub async fn delete(pool: &DbPool, instance_id: InstanceId) -> Result<usize, Error> {
  50     let conn = &mut get_conn(pool).await?;
  51     diesel::delete(instance::table.find(instance_id))
  52       .execute(conn)
  53       .await
  54   }
  55   #[cfg(test)]
  56   pub async fn delete_all(pool: &DbPool) -> Result<usize, Error> {
  57     let conn = &mut get_conn(pool).await?;
  58     diesel::delete(instance::table).execute(conn).await
  59   }
  60   pub async fn allowlist(pool: &DbPool) -> Result<Vec<Self>, Error> {
  61     let conn = &mut get_conn(pool).await?;
  62     instance::table
  63       .inner_join(federation_allowlist::table)
  64       .select(instance::all_columns)
  65       .get_results(conn)
  66       .await
  67   }
  68
  69   pub async fn blocklist(pool: &DbPool) -> Result<Vec<Self>, Error> {
  70     let conn = &mut get_conn(pool).await?;
  71     instance::table
  72       .inner_join(federation_blocklist::table)
  73       .select(instance::all_columns)
  74       .get_results(conn)
  75       .await
  76   }
  77
  78   pub async fn linked(pool: &DbPool) -> Result<Vec<Self>, Error> {
  79     let conn = &mut get_conn(pool).await?;
  80     instance::table
  81       .left_join(federation_blocklist::table)
  82       .filter(federation_blocklist::id.is_null())
  83       .select(instance::all_columns)
  84       .get_results(conn)
  85       .await
  86   }
  87 }