database: Implement CloudFrontInvalidationQueueItem models

Author: Turbo87

crates/crates_io_database/src/models/cloudfront_invalidation_queue.rs

@@ -0,0 +1,74 @@
+use crate::schema::cloudfront_invalidation_queue;
+use diesel::dsl::{count, exists};
+use diesel::prelude::*;
+use diesel_async::{AsyncPgConnection, RunQueryDsl};
+
+#[derive(Debug, Identifiable, Queryable, QueryableByName, Selectable)]
+#[diesel(table_name = cloudfront_invalidation_queue, check_for_backend(diesel::pg::Pg))]
+pub struct CloudFrontInvalidationQueueItem {
+    pub id: i64,
+    pub path: String,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Debug, Insertable)]
+#[diesel(table_name = cloudfront_invalidation_queue, check_for_backend(diesel::pg::Pg))]
+pub struct NewCloudFrontInvalidationQueueItem<'a> {
+    pub path: &'a str,
+}
+
+impl CloudFrontInvalidationQueueItem {
+    /// Queue multiple invalidation paths for later processing
+    pub async fn queue_paths(conn: &mut AsyncPgConnection, paths: &[String]) -> QueryResult<usize> {
+        let new_items: Vec<_> = paths
+            .iter()
+            .map(|path| NewCloudFrontInvalidationQueueItem { path })
+            .collect();
+
+        diesel::insert_into(cloudfront_invalidation_queue::table)
+            .values(&new_items)
+            .execute(conn)
+            .await
+    }
+
+    /// Fetch the oldest paths from the queue
+    pub async fn fetch_batch(
+        conn: &mut AsyncPgConnection,
+        limit: i64,
+    ) -> QueryResult<Vec<CloudFrontInvalidationQueueItem>> {
+        // Fetch the oldest entries up to the limit
+        cloudfront_invalidation_queue::table
+            .order(cloudfront_invalidation_queue::created_at.asc())
+            .limit(limit)
+            .load(conn)
+            .await
+    }
+
+    /// Remove queue items by their IDs
+    pub async fn remove_items(
+        conn: &mut AsyncPgConnection,
+        item_ids: &[i64],
+    ) -> QueryResult<usize> {
+        diesel::delete(cloudfront_invalidation_queue::table)
+            .filter(cloudfront_invalidation_queue::id.eq_any(item_ids))
+            .execute(conn)
+            .await
+    }
+
+    /// Check if there are more items in the queue
+    pub async fn has_pending_items(conn: &mut AsyncPgConnection) -> QueryResult<bool> {
+        diesel::select(exists(
+            cloudfront_invalidation_queue::table.select(cloudfront_invalidation_queue::id),
+        ))
+        .get_result(conn)
+        .await
+    }
+
+    /// Get current queue depth for monitoring
+    pub async fn get_queue_depth(conn: &mut AsyncPgConnection) -> QueryResult<i64> {
+        cloudfront_invalidation_queue::table
+            .select(count(cloudfront_invalidation_queue::id))
+            .first(conn)
+            .await
+    }
+}

crates/crates_io_database/src/models/mod.rs

@@ -1,5 +1,6 @@
 pub use self::action::{NewVersionOwnerAction, VersionAction, VersionOwnerAction};
 pub use self::category::{Category, CrateCategory, NewCategory};
+pub use self::cloudfront_invalidation_queue::CloudFrontInvalidationQueueItem;
 pub use self::crate_owner_invitation::{
     CrateOwnerInvitation, NewCrateOwnerInvitation, NewCrateOwnerInvitationOutcome,
 };
@@ -22,6 +23,7 @@ pub mod helpers;
 
 mod action;
 pub mod category;
+mod cloudfront_invalidation_queue;
 pub mod crate_owner_invitation;
 pub mod default_versions;
 mod deleted_crate;