Change how cleanup trait fn works using a date cutoff

api/adaptors/README.md

@@ -10,6 +10,7 @@ Note, you will need to have the following crates as dependencies in your adaptor
 
 - `common`<br>Includes a trait for implementing your adaptor, as well as structs your adaptor needs to return.
 - `async-trait`<br>Required because the trait from `common` uses async functions, make sure you include `#[async_trait]` above your trait implementation.
+- `chrono`<br>Required to deal with dates in the common structs and trait function signatures.
 
 Once you've created the adaptor, you'll need to make sure it's included as a dependency in the root [`Cargo.toml`](../Cargo.toml), and add a feature flag with the same name. Make sure you also document the new adaptor in the [api readme](../README.md).
 

api/adaptors/datastore/src/lib.rs

@@ -2,15 +2,10 @@ use std::{env, error::Error, fmt::Display};
 
 use async_trait::async_trait;
 use chrono::{DateTime, NaiveDateTime, Utc};
-use common::{
-    adaptor::Adaptor,
-    event::{Event, EventDeletion},
-    person::Person,
-    stats::Stats,
-};
+use common::{Adaptor, Event, Person, Stats};
 use google_cloud::{
     authorize::ApplicationCredentials,
-    datastore::{Client, Filter, FromValue, IntoValue, Key, Query},
+    datastore::{Client, Filter, FromValue, IntoValue, Key, KeyID, Query},
 };
 use tokio::sync::Mutex;
 
@@ -95,9 +90,22 @@ impl Adaptor for DatastoreAdaptor {
         ))
     }
 
-    async fn upsert_person(&self, event_id: String, person: Person) -> Result<Person, Self::Error> {
+    async fn upsert_person(
+        &self,
+        event_id: String,
+        person: Person,
+    ) -> Result<Option<Person>, Self::Error> {
         let mut client = self.client.lock().await;
 
+        // Check the event exists
+        if client
+            .get::<DatastoreEvent, _>(Key::new(EVENT_KIND).id(event_id.clone()))
+            .await?
+            .is_none()
+        {
+            return Ok(None);
+        }
+
         // Check if person exists
         let existing_person = client
             .query(
@@ -122,7 +130,7 @@ impl Adaptor for DatastoreAdaptor {
             .put((key, DatastorePerson::from_person(person.clone(), event_id)))
             .await?;
 
-        Ok(person)
+        Ok(Some(person))
     }
 
     async fn get_event(&self, id: String) -> Result<Option<Event>, Self::Error> {
@@ -151,25 +159,45 @@ impl Adaptor for DatastoreAdaptor {
         Ok(event)
     }
 
-    async fn delete_event(&self, id: String) -> Result<EventDeletion, Self::Error> {
+    async fn delete_events(&self, cutoff: DateTime<Utc>) -> Result<Stats, Self::Error> {
         let mut client = self.client.lock().await;
 
         let mut keys_to_delete: Vec<Key> = client
-            .query(
-                Query::new(PERSON_KIND)
-                    .filter(Filter::Equal("eventId".into(), id.clone().into_value())),
-            )
+            .query(Query::new(EVENT_KIND).filter(Filter::LesserThan(
+                "visited".into(),
+                cutoff.timestamp().into_value(),
+            )))
             .await?
             .iter()
             .map(|entity| entity.key().clone())
             .collect();
 
-        let person_count = keys_to_delete.len().try_into().unwrap();
-        keys_to_delete.insert(0, Key::new(EVENT_KIND).id(id.clone()));
+        let event_count = keys_to_delete.len() as i64;
+
+        let events_to_delete = keys_to_delete.clone();
+        for e in events_to_delete.iter() {
+            if let KeyID::StringID(id) = e.get_id() {
+                let mut event_people_to_delete: Vec<Key> = client
+                    .query(
+                        Query::new(PERSON_KIND)
+                            .filter(Filter::Equal("eventId".into(), id.clone().into_value())),
+                    )
+                    .await?
+                    .iter()
+                    .map(|entity| entity.key().clone())
+                    .collect();
+                keys_to_delete.append(&mut event_people_to_delete);
+            }
+        }
+
+        let person_count = keys_to_delete.len() as i64 - event_count;
 
         client.delete_all(keys_to_delete).await?;
 
-        Ok(EventDeletion { id, person_count })
+        Ok(Stats {
+            event_count,
+            person_count,
+        })
     }
 }
 

api/adaptors/memory/src/lib.rs

@@ -1,13 +1,8 @@
 use std::{collections::HashMap, error::Error, fmt::Display};
 
 use async_trait::async_trait;
-use chrono::Utc;
-use common::{
-    adaptor::Adaptor,
-    event::{Event, EventDeletion},
-    person::Person,
-    stats::Stats,
-};
+use chrono::{DateTime, Utc};
+use common::{Adaptor, Event, Person, Stats};
 use tokio::sync::Mutex;
 
 struct State {
@@ -68,14 +63,23 @@ impl Adaptor for MemoryAdaptor {
         ))
     }
 
-    async fn upsert_person(&self, event_id: String, person: Person) -> Result<Person, Self::Error> {
+    async fn upsert_person(
+        &self,
+        event_id: String,
+        person: Person,
+    ) -> Result<Option<Person>, Self::Error> {
         let mut state = self.state.lock().await;
 
+        // Check event exists
+        if state.events.get(&event_id).is_none() {
+            return Ok(None);
+        }
+
         state
             .people
             .insert((event_id, person.name.clone()), person.clone());
 
-        Ok(person)
+        Ok(Some(person))
     }
 
     async fn get_event(&self, id: String) -> Result<Option<Event>, Self::Error> {
@@ -98,21 +102,38 @@ impl Adaptor for MemoryAdaptor {
         Ok(event)
     }
 
-    async fn delete_event(&self, id: String) -> Result<EventDeletion, Self::Error> {
+    async fn delete_events(&self, cutoff: DateTime<Utc>) -> Result<Stats, Self::Error> {
         let mut state = self.state.lock().await;
 
-        let mut person_count: u64 = state.people.len() as u64;
+        // Delete events older than cutoff date
+        let mut deleted_event_ids: Vec<String> = Vec::new();
+        state.events = state
+            .events
+            .clone()
+            .into_iter()
+            .filter(|(id, event)| {
+                if event.visited_at >= cutoff {
+                    true
+                } else {
+                    deleted_event_ids.push(id.into());
+                    false
+                }
+            })
+            .collect();
+
+        let mut person_count = state.people.len() as i64;
         state.people = state
             .people
             .clone()
             .into_iter()
-            .filter(|((event_id, _), _)| event_id != &id)
+            .filter(|((event_id, _), _)| deleted_event_ids.contains(event_id))
             .collect();
-        person_count -= state.people.len() as u64;
+        person_count -= state.people.len() as i64;
 
-        state.events.remove(&id);
-
-        Ok(EventDeletion { id, person_count })
+        Ok(Stats {
+            event_count: deleted_event_ids.len() as i64,
+            person_count,
+        })
     }
 }
 

api/adaptors/sql/src/lib.rs

@@ -1,13 +1,8 @@
 use std::{env, error::Error};
 
 use async_trait::async_trait;
-use chrono::{DateTime as ChronoDateTime, Utc};
-use common::{
-    adaptor::Adaptor,
-    event::{Event, EventDeletion},
-    person::Person,
-    stats::Stats,
-};
+use chrono::{DateTime, Utc};
+use common::{Adaptor, Event, Person, Stats};
 use entity::{event, person, stats};
 use migration::{Migrator, MigratorTrait};
 use sea_orm::{
@@ -70,7 +65,11 @@ impl Adaptor for SqlAdaptor {
         })
     }
 
-    async fn upsert_person(&self, event_id: String, person: Person) -> Result<Person, Self::Error> {
+    async fn upsert_person(
+        &self,
+        event_id: String,
+        person: Person,
+    ) -> Result<Option<Person>, Self::Error> {
         let data = person::ActiveModel {
             name: Set(person.name.clone()),
             password_hash: Set(person.password_hash),
@@ -79,7 +78,16 @@ impl Adaptor for SqlAdaptor {
             event_id: Set(event_id.clone()),
         };
 
-        Ok(
+        // Check if the event exists
+        if event::Entity::find_by_id(event_id.clone())
+            .one(&self.db)
+            .await?
+            .is_none()
+        {
+            return Ok(None);
+        }
+
+        Ok(Some(
             match person::Entity::find_by_id((person.name, event_id))
                 .one(&self.db)
                 .await?
@@ -87,7 +95,7 @@ impl Adaptor for SqlAdaptor {
                 Some(_) => data.update(&self.db).await?.try_into_model()?.into(),
                 None => data.insert(&self.db).await?.try_into_model()?.into(),
             },
-        )
+        ))
     }
 
     async fn get_event(&self, id: String) -> Result<Option<Event>, Self::Error> {
@@ -118,27 +126,43 @@ impl Adaptor for SqlAdaptor {
         .into())
     }
 
-    async fn delete_event(&self, id: String) -> Result<EventDeletion, Self::Error> {
-        let event_id = id.clone();
-        let person_count = self
+    async fn delete_events(&self, cutoff: DateTime<Utc>) -> Result<Stats, Self::Error> {
+        let (event_count, person_count) = self
             .db
-            .transaction::<_, u64, DbErr>(|t| {
+            .transaction::<_, (i64, i64), DbErr>(|t| {
                 Box::pin(async move {
+                    // Get events older than the cutoff date
+                    let old_events = event::Entity::find()
+                        .filter(event::Column::VisitedAt.lt(cutoff.naive_utc()))
+                        .all(t)
+                        .await?;
+
                     // Delete people
-                    let people_delete_result = person::Entity::delete_many()
-                        .filter(person::Column::EventId.eq(&event_id))
+                    let mut people_deleted: i64 = 0;
+                    // TODO: run concurrently
+                    for e in old_events.iter() {
+                        let people_delete_result = person::Entity::delete_many()
+                            .filter(person::Column::EventId.eq(&e.id))
+                            .exec(t)
+                            .await?;
+                        people_deleted += people_delete_result.rows_affected as i64;
+                    }
+
+                    // Delete events
+                    let event_delete_result = event::Entity::delete_many()
+                        .filter(event::Column::VisitedAt.lt(cutoff.naive_utc()))
                         .exec(t)
                         .await?;
 
-                    // Delete event
-                    event::Entity::delete_by_id(event_id).exec(t).await?;
-
-                    Ok(people_delete_result.rows_affected)
+                    Ok((event_delete_result.rows_affected as i64, people_deleted))
                 })
             })
             .await?;
 
-        Ok(EventDeletion { id, person_count })
+        Ok(Stats {
+            event_count,
+            person_count,
+        })
     }
 }
 
@@ -190,8 +214,8 @@ impl From<event::Model> for Event {
         Self {
             id: value.id,
             name: value.name,
-            created_at: ChronoDateTime::<Utc>::from_utc(value.created_at, Utc),
-            visited_at: ChronoDateTime::<Utc>::from_utc(value.visited_at, Utc),
+            created_at: DateTime::<Utc>::from_utc(value.created_at, Utc),
+            visited_at: DateTime::<Utc>::from_utc(value.visited_at, Utc),
             times: serde_json::from_value(value.times).unwrap_or(vec![]),
             timezone: value.timezone,
         }
@@ -203,7 +227,7 @@ impl From<person::Model> for Person {
         Self {
             name: value.name,
             password_hash: value.password_hash,
-            created_at: ChronoDateTime::<Utc>::from_utc(value.created_at, Utc),
+            created_at: DateTime::<Utc>::from_utc(value.created_at, Utc),
             availability: serde_json::from_value(value.availability).unwrap_or(vec![]),
         }
     }