diff --git a/Cargo.toml b/Cargo.toml index 330cbbe..440adc1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -71,6 +71,12 @@ conf-files = [ "/etc/reminder-rs/Rocket.toml", ] -[package.metadata.deb.systemd-units] +[[package.metadata.deb.systemd-units]] unit-scripts = "systemd" +unit-name = "reminder-rs" +start = false + +[[package.metadata.deb.systemd-units]] +unit-scripts = "systemd" +unit-name = "reminder-rs-clean-images" start = false diff --git a/README.md b/README.md index f759bae..4a09c31 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,27 @@ Image URL inputs offer an **Upload image** button when all of these settings are of NSFW images. Set it to `false` to skip NSFW moderation and model initialization. Empty or unrecognized values keep checks enabled. +The Debian package includes `reminder-rs-clean-images.timer`, an hourly cleanup job. Enable it +after configuring S3 and starting the bot (which applies database migrations): + +```sh +sudo systemctl enable --now reminder-rs-clean-images.timer +``` + +Each run records the first observed orphan time in `image_uploads.orphaned_at` (UTC). An upload +is orphaned when neither a reminder (including archived reminders) nor a template uses its URL +as an avatar, image, thumbnail, author icon or footer icon. Referenced uploads have this timestamp +cleared. `IMAGE_UPLOAD_ORPHAN_RETENTION_SECONDS` controls the additional retention time after +detection; it defaults to `604800` (seven days) and must be a positive integer. Unsaved uploads +are also covered. Cleanup rechecks references before deleting and removes the tracking row only +after S3 confirms deletion; failed deletions are retried on later runs. Each run attempts up to +1,000 expired uploads from the configured public URL's `uploads/` prefix. + +For other deployments, schedule `/path/to/reminder-rs clean-images` hourly with cron, using the +same working directory and configuration as the bot. The command requires no Discord token +and skips NSFW model initialization. With no S3 settings it exits successfully without work; +partial S3 settings or invalid retention values fail the job. + ## Running with Docker ### RustFS for local development diff --git a/conf/default.env b/conf/default.env index 864e5bf..b517fbb 100644 --- a/conf/default.env +++ b/conf/default.env @@ -32,3 +32,6 @@ S3_PUBLIC_URL= S3_ENDPOINT= # NSFW moderation is enabled by default. Set false (or 0/off) to disable it. NSFW_CHECK_ENABLED=true +# Seconds to retain an upload after cleanup first detects that it is orphaned. +# Must be positive. Defaults to seven days when absent or blank. +IMAGE_UPLOAD_ORPHAN_RETENTION_SECONDS=604800 diff --git a/migrations/20261012120000_image_upload_orphans.sql b/migrations/20261012120000_image_upload_orphans.sql new file mode 100644 index 0000000..0a3678a --- /dev/null +++ b/migrations/20261012120000_image_upload_orphans.sql @@ -0,0 +1,4 @@ +ALTER TABLE image_uploads + -- UTC time at which cleanup first observed that this upload had no references. + ADD COLUMN orphaned_at DATETIME(6) NULL DEFAULT NULL, + ADD INDEX image_uploads_orphaned_at (orphaned_at); diff --git a/src/image_cleanup.rs b/src/image_cleanup.rs new file mode 100644 index 0000000..98ae27b --- /dev/null +++ b/src/image_cleanup.rs @@ -0,0 +1,165 @@ +use std::env; + +use log::{info, warn}; +use sqlx::{Acquire, MySql, Pool}; + +use crate::{web::image_uploads::ImageUploadService, Error}; + +const DEFAULT_RETENTION_SECONDS: u32 = 7 * 24 * 60 * 60; +const BATCH_SIZE: u32 = 1000; +const IMAGE_COLUMNS: &str = + "avatar, embed_image_url, embed_thumbnail_url, embed_author_url, embed_footer_url"; + +fn binary_image_columns() -> String { + // Existing URL columns and image_uploads use different collations. S3 URLs must + // also match case-sensitively, so compare their bytes rather than text collation. + IMAGE_COLUMNS + .split(", ") + .map(|column| format!("BINARY {column}")) + .collect::>() + .join(", ") +} + +fn retention_seconds(value: Option<&str>) -> Result { + match value.map(str::trim).filter(|value| !value.is_empty()) { + None => Ok(DEFAULT_RETENTION_SECONDS), + Some(value) => value.parse::().ok().filter(|value| *value > 0).ok_or_else(|| { + "IMAGE_UPLOAD_ORPHAN_RETENTION_SECONDS must be a positive integer of seconds".into() + }), + } +} + +// Only keys generated by the upload route in the configured bucket may be deleted. +fn object_key<'a>(public_url: &str, url: &'a str) -> Option<&'a str> { + let key = url.strip_prefix(public_url)?.strip_prefix('/')?; + let name = key.strip_prefix("uploads/")?.strip_suffix(".png")?; + (name.len() == 64 + && name.bytes().all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))) + .then_some(key) +} + +async fn mark_orphans(pool: &Pool) -> Result<(), Error> { + let columns = binary_image_columns(); + sqlx::query(&format!( + "UPDATE image_uploads AS upload SET orphaned_at = CASE + WHEN NOT EXISTS (SELECT 1 FROM reminders + WHERE BINARY upload.url IN ({columns})) + AND NOT EXISTS (SELECT 1 FROM reminder_template + WHERE BINARY upload.url IN ({columns})) + THEN COALESCE(upload.orphaned_at, UTC_TIMESTAMP(6)) ELSE NULL END" + )) + .execute(pool) + .await?; + Ok(()) +} + +pub(crate) async fn run(pool: &Pool) -> Result<(), Error> { + let retention = + retention_seconds(env::var("IMAGE_UPLOAD_ORPHAN_RETENTION_SECONDS").ok().as_deref())?; + let Some(storage) = ImageUploadService::for_cleanup()? else { + info!("Image cleanup skipped: S3 uploads are not configured"); + return Ok(()); + }; + cleanup(pool, &storage, retention).await +} + +async fn cleanup( + pool: &Pool, + storage: &ImageUploadService, + retention: u32, +) -> Result<(), Error> { + mark_orphans(pool).await?; + let candidates: Vec = sqlx::query_scalar( + "SELECT id FROM image_uploads + WHERE orphaned_at <= UTC_TIMESTAMP(6) - INTERVAL ? SECOND + ORDER BY orphaned_at, id LIMIT ?", + ) + .bind(retention) + .bind(BATCH_SIZE) + .fetch_all(pool) + .await?; + + let mut deleted = 0; + let mut failed = 0; + for id in candidates { + match delete_orphan(pool, storage, retention, id).await { + Ok(true) => deleted += 1, + Ok(false) => {} + Err(error) => { + failed += 1; + warn!("Could not clean up image upload {}: {}", id, error); + } + } + } + info!("Image cleanup deleted {} uploads; {} failed", deleted, failed); + if failed > 0 { + return Err(format!("Image cleanup failed for {failed} uploads; retained for retry").into()); + } + Ok(()) +} + +async fn delete_orphan( + pool: &Pool, + storage: &ImageUploadService, + retention: u32, + id: u64, +) -> Result { + let mut connection = pool.acquire().await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") + .execute(&mut *connection) + .await?; + let mut transaction = connection.begin().await?; + // Serialize overlapping cleanup jobs and recheck the retention time under the row lock. + let url: Option = sqlx::query_scalar( + "SELECT url FROM image_uploads + WHERE id = ? AND orphaned_at <= UTC_TIMESTAMP(6) - INTERVAL ? SECOND + FOR UPDATE", + ) + .bind(id) + .bind(retention) + .fetch_optional(&mut *transaction) + .await?; + let Some(url) = url else { + transaction.rollback().await?; + return Ok(false); + }; + let key = object_key(&storage.public_url, &url) + .ok_or("Upload URL does not match a generated key in the configured S3 public URL")?; + + for table in ["reminders", "reminder_template"] { + // Locking reads see current committed references, rather than a stale snapshot. + // InnoDB's REPEATABLE READ next-key locks also prevent a reference from + // being inserted while the object is being removed. Keep this transaction short: + // it covers one S3 request with the upload service's 30-second timeout. + let columns = binary_image_columns(); + let referenced: Option = sqlx::query_scalar(&format!( + "SELECT id FROM {table} WHERE BINARY ? IN ({columns}) LIMIT 1 FOR UPDATE" + )) + .bind(&url) + .fetch_optional(&mut *transaction) + .await?; + if referenced.is_some() { + sqlx::query("UPDATE image_uploads SET orphaned_at = NULL WHERE id = ?") + .bind(id) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + return Ok(false); + } + } + + let response = storage.bucket.delete_object(key).await?; + // A missing object is already cleaned up, including a retry after a DB commit failure. + if !(200..300).contains(&response.status_code()) && response.status_code() != 404 { + return Err(format!("S3 deletion returned HTTP {}", response.status_code()).into()); + } + sqlx::query("DELETE FROM image_uploads WHERE id = ?") + .bind(id) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + Ok(true) +} + +#[cfg(test)] +mod tests; diff --git a/src/image_cleanup/tests.rs b/src/image_cleanup/tests.rs new file mode 100644 index 0000000..f36c882 --- /dev/null +++ b/src/image_cleanup/tests.rs @@ -0,0 +1,258 @@ +use std::{sync::Arc, time::Duration}; + +use chrono::NaiveDateTime; +use s3::{creds::Credentials, Bucket, Region}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + sync::Semaphore, + task::JoinHandle, +}; + +use super::*; + +const BASE_URL: &str = "https://images.example.com/bucket"; + +fn upload_url(number: u64) -> String { + format!("{BASE_URL}/uploads/{number:064x}.png") +} + +async fn upload(pool: &Pool, number: u64) -> u64 { + sqlx::query("INSERT INTO image_uploads (uploader, url) VALUES (123, ?)") + .bind(upload_url(number)) + .execute(pool) + .await + .unwrap() + .last_insert_id() +} + +async fn orphaned_at(pool: &Pool, id: u64) -> Option { + sqlx::query_scalar("SELECT orphaned_at FROM image_uploads WHERE id = ?") + .bind(id) + .fetch_one(pool) + .await + .unwrap() +} + +async fn expire(pool: &Pool, id: u64) { + sqlx::query( + "UPDATE image_uploads SET orphaned_at = UTC_TIMESTAMP(6) - INTERVAL 2 DAY WHERE id = ?", + ) + .bind(id) + .execute(pool) + .await + .unwrap(); +} + +async fn reference_tables(pool: &Pool) -> (u64, u64) { + let guild = sqlx::query("INSERT INTO guilds (guild) VALUES (123)") + .execute(pool) + .await + .unwrap() + .last_insert_id(); + let channel = sqlx::query("INSERT INTO channels (channel, guild_id) VALUES (123, ?)") + .bind(guild) + .execute(pool) + .await + .unwrap() + .last_insert_id(); + (guild, channel) +} + +async fn storage(status: u16) -> (ImageUploadService, JoinHandle) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + let (mut socket, _) = + tokio::time::timeout(Duration::from_secs(5), listener.accept()).await.unwrap().unwrap(); + let mut received = Vec::new(); + let mut buffer = [0; 4096]; + while !received.windows(4).any(|window| window == b"\r\n\r\n") { + let count = socket.read(&mut buffer).await.unwrap(); + assert!(count > 0); + received.extend_from_slice(&buffer[..count]); + } + socket + .write_all( + format!("HTTP/1.1 {status} Test\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .as_bytes(), + ) + .await + .unwrap(); + String::from_utf8(received).unwrap() + }); + let credentials = Credentials::new(Some("test"), Some("test"), None, None, None).unwrap(); + let mut bucket = + Bucket::new("images", Region::Custom { region: "local".into(), endpoint }, credentials) + .unwrap() + .with_path_style(); + bucket.set_request_timeout(Some(Duration::from_secs(2))); + ( + ImageUploadService { + bucket, + public_url: BASE_URL.into(), + model: None, + slots: Arc::new(Semaphore::new(2)), + }, + task, + ) +} + +#[test] +fn retention_defaults_and_rejects_invalid_settings() { + for value in [None, Some(""), Some(" ")] { + assert_eq!(retention_seconds(value).unwrap(), 604800); + } + assert_eq!(retention_seconds(Some(" 3600 ")).unwrap(), 3600); + for value in ["0", "-1", "1.5", "forever", "4294967296"] { + assert!(retention_seconds(Some(value)).is_err()); + } +} + +#[test] +fn restricts_deletion_to_generated_keys_in_the_configured_url() { + assert_eq!( + object_key(BASE_URL, &upload_url(1)), + Some(format!("uploads/{:064x}.png", 1).as_str()) + ); + for url in [ + upload_url(1).replace("images.example.com", "other.example.com"), + upload_url(1).replace("/bucket/", "/bucket-other/"), + upload_url(1).replace("/uploads/", "/private/"), + format!("{BASE_URL}/uploads/../private.png"), + format!("{BASE_URL}/uploads/{}.png", "g".repeat(64)), + format!("{}?download=true", upload_url(1)), + format!("{BASE_URL}/uploads/short.png"), + ] { + assert!(object_key(BASE_URL, &url).is_none(), "{url}"); + } +} + +#[sqlx::test] +async fn orphan_timestamp_survives_repeated_runs_and_resets_on_reuse(pool: Pool) { + let id = upload(&pool, 1).await; + assert!(orphaned_at(&pool, id).await.is_none()); + mark_orphans(&pool).await.unwrap(); + let first = orphaned_at(&pool, id).await.unwrap(); + mark_orphans(&pool).await.unwrap(); + assert_eq!(orphaned_at(&pool, id).await, Some(first)); + + let (guild, _) = reference_tables(&pool).await; + let template = + sqlx::query("INSERT INTO reminder_template (guild_id, embed_image_url) VALUES (?, ?)") + .bind(guild) + .bind(upload_url(1)) + .execute(&pool) + .await + .unwrap() + .last_insert_id(); + mark_orphans(&pool).await.unwrap(); + assert!(orphaned_at(&pool, id).await.is_none()); + sqlx::query("DELETE FROM reminder_template WHERE id = ?") + .bind(template) + .execute(&pool) + .await + .unwrap(); + mark_orphans(&pool).await.unwrap(); + assert!(orphaned_at(&pool, id).await.unwrap() > first); +} + +#[sqlx::test] +async fn protects_all_image_fields_in_reminders_and_templates(pool: Pool) { + let (guild, channel) = reference_tables(&pool).await; + let mut ids = Vec::new(); + for (index, column) in IMAGE_COLUMNS.split(", ").enumerate() { + for table in ["reminders", "reminder_template"] { + let number = ids.len() as u64 + 1; + ids.push(upload(&pool, number).await); + let query = if table == "reminders" { + // Archived/disabled reminders still retain their image URLs. + format!( + "INSERT INTO reminders (uid, channel_id, `utc_time`, status, enabled, {column}) + VALUES ('field-{index}', {channel}, UTC_TIMESTAMP(), 'sent', 0, ?)" + ) + } else { + format!("INSERT INTO reminder_template (guild_id, {column}) VALUES ({guild}, ?)") + }; + sqlx::query(&query).bind(upload_url(number)).execute(&pool).await.unwrap(); + } + } + let unused = upload(&pool, 100).await; + // Previously orphaned files are unmarked as soon as cleanup observes a reference. + for id in &ids { + expire(&pool, *id).await; + } + mark_orphans(&pool).await.unwrap(); + for id in ids { + assert!(orphaned_at(&pool, id).await.is_none()); + } + assert!(orphaned_at(&pool, unused).await.is_some()); +} + +#[sqlx::test] +async fn only_expired_orphans_are_deleted_and_rows_removed_after_success(pool: Pool) { + let old = upload(&pool, 1).await; + let fresh = upload(&pool, 2).await; + expire(&pool, old).await; + let (storage, request) = storage(204).await; + cleanup(&pool, &storage, 86400).await.unwrap(); + assert!(request.await.unwrap().starts_with(&format!("DELETE /images/uploads/{:064x}.png ", 1))); + let ids: Vec = sqlx::query_scalar("SELECT id FROM image_uploads ORDER BY id") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(ids, vec![fresh]); + assert!(orphaned_at(&pool, fresh).await.is_some()); +} + +#[sqlx::test] +async fn failed_deletions_keep_the_original_timestamp_and_retry_missing_objects(pool: Pool) { + let id = upload(&pool, 1).await; + expire(&pool, id).await; + let original = orphaned_at(&pool, id).await; + let (failed_storage, failed_request) = storage(403).await; + assert!(cleanup(&pool, &failed_storage, 86400).await.is_err()); + failed_request.await.unwrap(); + assert_eq!(orphaned_at(&pool, id).await, original); + + let (retry_storage, retry_request) = storage(404).await; + cleanup(&pool, &retry_storage, 86400).await.unwrap(); + retry_request.await.unwrap(); + let count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM image_uploads").fetch_one(&pool).await.unwrap(); + assert_eq!(count, 0); +} + +#[sqlx::test] +async fn rechecks_references_before_deleting_a_candidate(pool: Pool) { + let id = upload(&pool, 1).await; + expire(&pool, id).await; + let (guild, _) = reference_tables(&pool).await; + sqlx::query("INSERT INTO reminder_template (guild_id, avatar) VALUES (?, ?)") + .bind(guild) + .bind(upload_url(1)) + .execute(&pool) + .await + .unwrap(); + let (storage, request) = storage(204).await; + assert!(!delete_orphan(&pool, &storage, 86400, id).await.unwrap()); + assert!(orphaned_at(&pool, id).await.is_none()); + request.abort(); +} + +#[sqlx::test] +async fn foreign_urls_are_retained_without_contacting_s3(pool: Pool) { + let id = upload(&pool, 1).await; + sqlx::query( + "UPDATE image_uploads SET url = 'https://elsewhere.example.com/file.png' WHERE id = ?", + ) + .bind(id) + .execute(&pool) + .await + .unwrap(); + expire(&pool, id).await; + let (storage, request) = storage(204).await; + assert!(cleanup(&pool, &storage, 86400).await.is_err()); + assert!(orphaned_at(&pool, id).await.is_some()); + request.abort(); +} diff --git a/src/main.rs b/src/main.rs index 48ae1c8..b971d5a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,6 +9,7 @@ mod consts; mod event_handlers; #[cfg(not(test))] mod hooks; +mod image_cleanup; mod interval_parser; mod metrics; #[cfg(not(test))] @@ -121,6 +122,7 @@ async fn _main(tx: Sender<()>) -> Result<(), Box> { Pool::connect(&env::var("DATABASE_URL").expect("No database URL provided")).await.unwrap(); match cmd_word { + Some("clean-images") => image_cleanup::run(&database).await, Some("clean") => { let sent_clean_age = env::var("SENT_CLEAN_AGE").expect("No SENT_CLEAN_AGE provided"); if sent_clean_age.is_empty() { diff --git a/src/web/image_uploads.rs b/src/web/image_uploads.rs index e290b82..327b24e 100644 --- a/src/web/image_uploads.rs +++ b/src/web/image_uploads.rs @@ -57,6 +57,20 @@ impl ImageUploads { } impl ImageUploadService { + // Cleanup uses the same storage configuration without loading the moderation model. + pub(crate) fn for_cleanup() -> Result, Box> { + let values = + ["S3_BUCKET", "S3_REGION", "S3_ACCESS_KEY_ID", "S3_SECRET_ACCESS_KEY", "S3_PUBLIC_URL"] + .map(|name| env::var(name).unwrap_or_default()); + if values.iter().all(|value| value.trim().is_empty()) { + return Ok(None); + } + if values.iter().any(|value| value.trim().is_empty()) { + return Err("Image cleanup requires all S3 upload settings".into()); + } + Self::new(&values, false).map(Some).map_err(|error| error.to_string().into()) + } + fn new(values: &[String; 5], check_nsfw: bool) -> Result> { let url = reqwest::Url::parse(&values[4])?; let public_url = url.as_str().trim_end_matches('/').to_string(); diff --git a/src/web/mod.rs b/src/web/mod.rs index e0d4f92..4f67c8a 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -4,7 +4,7 @@ mod macros; mod catchers; mod fairings; mod guards; -mod image_uploads; +pub(crate) mod image_uploads; mod ip_blocking; mod routes; diff --git a/systemd/reminder-rs-clean-images.service b/systemd/reminder-rs-clean-images.service new file mode 100644 index 0000000..e62bcff --- /dev/null +++ b/systemd/reminder-rs-clean-images.service @@ -0,0 +1,11 @@ +[Unit] +Description=Clean orphaned image uploads from Reminder Bot +Wants=network-online.target +After=network-online.target + +[Service] +User=reminder +Type=oneshot +ExecStart=/usr/bin/reminder-rs clean-images +WorkingDirectory=/etc/reminder-rs +Environment="RUST_LOG=reminder_rs=info" diff --git a/systemd/reminder-rs-clean-images.timer b/systemd/reminder-rs-clean-images.timer new file mode 100644 index 0000000..cc51b1d --- /dev/null +++ b/systemd/reminder-rs-clean-images.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Clean orphaned Reminder Bot image uploads hourly + +[Timer] +OnCalendar=hourly +Persistent=true +Unit=reminder-rs-clean-images.service + +[Install] +WantedBy=timers.target