crates/db_schema/src/impls/activity.rs

   1 use crate::{
   2   diesel::OptionalExtension,
   3   newtypes::DbUrl,
   4   source::activity::{ReceivedActivity, SentActivity, SentActivityForm},
   5   utils::{get_conn, DbPool},
   6 };
   7 use diesel::{
   8   dsl::insert_into,
   9   result::{DatabaseErrorKind, Error, Error::DatabaseError},
  10   ExpressionMethods,
  11   QueryDsl,
  12 };
  13 use diesel_async::RunQueryDsl;
  14
  15 impl SentActivity {
  16   pub async fn create(pool: &mut DbPool<'_>, form: SentActivityForm) -> Result<Self, Error> {
  17     use crate::schema::sent_activity::dsl::sent_activity;
  18     let conn = &mut get_conn(pool).await?;
  19     insert_into(sent_activity)
  20       .values(form)
  21       .get_result::<Self>(conn)
  22       .await
  23   }
  24
  25   pub async fn read_from_apub_id(pool: &mut DbPool<'_>, object_id: &DbUrl) -> Result<Self, Error> {
  26     use crate::schema::sent_activity::dsl::{ap_id, sent_activity};
  27     let conn = &mut get_conn(pool).await?;
  28     sent_activity
  29       .filter(ap_id.eq(object_id))
  30       .first::<Self>(conn)
  31       .await
  32   }
  33 }
  34
  35 impl ReceivedActivity {
  36   pub async fn create(pool: &mut DbPool<'_>, ap_id_: &DbUrl) -> Result<(), Error> {
  37     use crate::schema::received_activity::dsl::{ap_id, id, received_activity};
  38     let conn = &mut get_conn(pool).await?;
  39     let res = insert_into(received_activity)
  40       .values(ap_id.eq(ap_id_))
  41       .on_conflict_do_nothing()
  42       .returning(id)
  43       .get_result::<i64>(conn)
  44       .await
  45       .optional()?;
  46     if res.is_some() {
  47       // new activity inserted successfully
  48       Ok(())
  49     } else {
  50       // duplicate activity
  51       Err(DatabaseError(
  52         DatabaseErrorKind::UniqueViolation,
  53         Box::<String>::default(),
  54       ))
  55     }
  56   }
  57 }
  58
  59 #[cfg(test)]
  60 mod tests {
  61   use super::*;
  62   use crate::utils::build_db_pool_for_tests;
  63   use serde_json::json;
  64   use serial_test::serial;
  65   use url::Url;
  66
  67   #[tokio::test]
  68   #[serial]
  69   async fn receive_activity_duplicate() {
  70     let pool = &build_db_pool_for_tests().await;
  71     let pool = &mut pool.into();
  72     let ap_id: DbUrl = Url::parse("http://example.com/activity/531")
  73       .unwrap()
  74       .into();
  75
  76     // inserting activity for first time
  77     let res = ReceivedActivity::create(pool, &ap_id).await;
  78     assert!(res.is_ok());
  79
  80     let res = ReceivedActivity::create(pool, &ap_id).await;
  81     assert!(res.is_err());
  82   }
  83
  84   #[tokio::test]
  85   #[serial]
  86   async fn sent_activity_write_read() {
  87     let pool = &build_db_pool_for_tests().await;
  88     let pool = &mut pool.into();
  89     let ap_id: DbUrl = Url::parse("http://example.com/activity/412")
  90       .unwrap()
  91       .into();
  92     let data = json!({
  93         "key1": "0xF9BA143B95FF6D82",
  94         "key2": "42",
  95     });
  96     let sensitive = false;
  97
  98     let form = SentActivityForm {
  99       ap_id: ap_id.clone(),
 100       data: data.clone(),
 101       sensitive,
 102     };
 103
 104     SentActivity::create(pool, form).await.unwrap();
 105
 106     let res = SentActivity::read_from_apub_id(pool, &ap_id).await.unwrap();
 107     assert_eq!(res.ap_id, ap_id);
 108     assert_eq!(res.data, data);
 109     assert_eq!(res.sensitive, sensitive);
 110   }
 111 }