Implement automatic pruning of remote posts

CHANGELOG.md

@@ -14,6 +14,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - Added `limits` parameter group to configuration.
 - Made file size limit adjustable with `limits.media.file_size_limit` configuration option.
 - Added `limits.posts.character_limit` configuration parameter (replaces `post_character_limit`).
+- Implemented automatic pruning of remote posts (disabled by default).
 
 ### Changed
 

contrib/mitra_config.yaml

@@ -35,6 +35,10 @@ registration:
 #  posts:
 #    character_limit: 2000
 
+# Data retention parameters
+#retention:
+#  extraneous_posts: 50
+
 # List of blocked domains
 #blocked_instances: []
 

src/config/main.rs

@@ -17,6 +17,7 @@ use crate::utils::urls::guess_protocol;
 use super::blockchain::BlockchainConfig;
 use super::environment::Environment;
 use super::limits::Limits;
+use super::retention::RetentionConfig;
 use super::MITRA_VERSION;
 
 #[derive(Clone, PartialEq)]
@@ -104,6 +105,9 @@ pub struct Config {
     #[serde(default)]
     pub limits: Limits,
 
+    #[serde(default)]
+    pub retention: RetentionConfig,
+
     #[serde(default)]
     pub blocked_instances: Vec<String>,
 

src/config/mod.rs

@@ -3,6 +3,7 @@ mod environment;
 mod limits;
 mod loader;
 mod main;
+mod retention;
 
 pub use blockchain::{
     BlockchainConfig,

src/config/retention.rs

@@ -0,0 +1,15 @@
+use serde::Deserialize;
+
+#[derive(Clone, Deserialize)]
+pub struct RetentionConfig {
+    pub extraneous_posts: Option<u32>,
+}
+
+#[allow(clippy::derivable_impls)]
+impl Default for RetentionConfig {
+    fn default() -> Self {
+        Self {
+            extraneous_posts: None,
+        }
+    }
+}

src/job_queue/periodic_tasks.rs

@@ -18,6 +18,8 @@ use crate::ethereum::subscriptions::{
     update_expired_subscriptions,
 };
 use crate::monero::subscriptions::check_monero_subscriptions;
+use crate::models::posts::queries::{delete_post, find_extraneous_posts};
+use crate::utils::datetime::days_before_now;
 
 pub async fn nft_monitor(
     maybe_blockchain: Option<&mut Blockchain>,
@@ -115,3 +117,21 @@ pub async fn outgoing_activity_queue_executor(
     process_queued_outgoing_activities(config, db_pool).await?;
     Ok(())
 }
+
+pub async fn delete_extraneous_posts(
+    config: &Config,
+    db_pool: &DbPool,
+) -> Result<(), Error> {
+    let db_client = &mut **get_database_client(db_pool).await?;
+    let updated_before = match config.retention.extraneous_posts {
+        Some(days) => days_before_now(days),
+        None => return Ok(()), // not configured
+    };
+    let posts = find_extraneous_posts(db_client, &updated_before).await?;
+    for post_id in posts {
+        let deletion_queue = delete_post(db_client, &post_id).await?;
+        deletion_queue.process(config).await;
+        log::info!("deleted post {}", post_id);
+    };
+    Ok(())
+}

src/job_queue/scheduler.rs

@@ -18,6 +18,7 @@ enum PeriodicTask {
     MoneroPaymentMonitor,
     IncomingActivityQueueExecutor,
     OutgoingActivityQueueExecutor,
+    DeleteExtraneousPosts,
 }
 
 impl PeriodicTask {
@@ -30,6 +31,7 @@ impl PeriodicTask {
             Self::MoneroPaymentMonitor => 30,
             Self::IncomingActivityQueueExecutor => 5,
             Self::OutgoingActivityQueueExecutor => 5,
+            Self::DeleteExtraneousPosts => 3600,
         }
     }
 
@@ -58,6 +60,9 @@ pub fn run(
             (PeriodicTask::IncomingActivityQueueExecutor, None),
             (PeriodicTask::OutgoingActivityQueueExecutor, None),
         ]);
+        if config.retention.extraneous_posts.is_some() {
+            scheduler_state.insert(PeriodicTask::DeleteExtraneousPosts, None);
+        };
 
         let mut interval = tokio::time::interval(Duration::from_secs(5));
         let mut token_waitlist_map: HashMap<Uuid, DateTime<Utc>> = HashMap::new();
@@ -95,6 +100,9 @@ pub fn run(
                     PeriodicTask::OutgoingActivityQueueExecutor => {
                         outgoing_activity_queue_executor(&config, &db_pool).await
                     },
+                    PeriodicTask::DeleteExtraneousPosts => {
+                        delete_extraneous_posts(&config, &db_pool).await
+                    },
                 };
                 task_result.unwrap_or_else(|err| {
                     log::error!("{:?}: {}", task, err);