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   #![allow(clippy::unwrap_used)]
  62   #![allow(clippy::indexing_slicing)]
  63
  64   use super::*;
  65   use crate::utils::build_db_pool_for_tests;
  66   use serde_json::json;
  67   use serial_test::serial;
  68   use url::Url;
  69
  70   #[tokio::test]
  71   #[serial]
  72   async fn receive_activity_duplicate() {
  73     let pool = &build_db_pool_for_tests().await;
  74     let pool = &mut pool.into();
  75     let ap_id: DbUrl = Url::parse("http://example.com/activity/531")
  76       .unwrap()
  77       .into();
  78
  79     // inserting activity for first time
  80     let res = ReceivedActivity::create(pool, &ap_id).await;
  81     assert!(res.is_ok());
  82
  83     let res = ReceivedActivity::create(pool, &ap_id).await;
  84     assert!(res.is_err());
  85   }
  86
  87   #[tokio::test]
  88   #[serial]
  89   async fn sent_activity_write_read() {
  90     let pool = &build_db_pool_for_tests().await;
  91     let pool = &mut pool.into();
  92     let ap_id: DbUrl = Url::parse("http://example.com/activity/412")
  93       .unwrap()
  94       .into();
  95     let data = json!({
  96         "key1": "0xF9BA143B95FF6D82",
  97         "key2": "42",
  98     });
  99     let sensitive = false;
 100
 101     let form = SentActivityForm {
 102       ap_id: ap_id.clone(),
 103       data: data.clone(),
 104       sensitive,
 105     };
 106
 107     SentActivity::create(pool, form).await.unwrap();
 108
 109     let res = SentActivity::read_from_apub_id(pool, &ap_id).await.unwrap();
 110     assert_eq!(res.ap_id, ap_id);
 111     assert_eq!(res.data, data);
 112     assert_eq!(res.sensitive, sensitive);
 113   }
 114 }