Create updates.xml RSS feed

This creates an RSS feed published at https://static.crates.io/rss/updates.xml. The feed is synced with the database in a background job after every successful publish of a new version. It includes the latest 100 published versions with the crate name, version number, crate description, URL and publish date.

Author: Turbo87

Cargo.lock

@@ -100,6 +100,7 @@ paste = "=1.0.15"
 prometheus = { version = "=0.13.4", default-features = false }
 rand = "=0.8.5"
 reqwest = { version = "=0.12.5", features = ["blocking", "gzip", "json"] }
+rss = { version = "=2.0.8", default-features = false, features = ["atom"] }
 scheduled-thread-pool = "=0.2.7"
 secrecy = "=0.8.0"
 semver = { version = "=1.0.23", features = ["serde"] }

Cargo.toml

@@ -39,6 +39,7 @@ pub enum Command {
         force: bool,
     },
     SendTokenExpiryNotifications,
+    SyncUpdatesFeed,
 }
 
 pub fn run(command: Command) -> Result<()> {
@@ -125,6 +126,9 @@ pub fn run(command: Command) -> Result<()> {
         Command::SendTokenExpiryNotifications => {
             jobs::SendTokenExpiryNotifications.enqueue(conn)?;
         }
+        Command::SyncUpdatesFeed => {
+            jobs::rss::SyncUpdatesFeed.enqueue(conn)?;
+        }
     };
 
     Ok(())

src/admin/enqueue_job.rs

@@ -431,6 +431,10 @@ pub async fn publish(app: AppState, req: BytesRequest) -> AppResult<Json<GoodCra
                 CheckTyposquat::new(&krate.name).enqueue(conn)?;
             }
 
+            if let Err(error) = jobs::rss::SyncUpdatesFeed.enqueue(conn) {
+                error!("Failed to enqueue `rss::SyncUpdatesFeed` job: {error}");
+            }
+
             // The `other` field on `PublishWarnings` was introduced to handle a temporary warning
             // that is no longer needed. As such, crates.io currently does not return any `other`
             // warnings at this time, but if we need to, the field is available.

src/controllers/krate/publish.rs

@@ -7,9 +7,10 @@ use object_store::local::LocalFileSystem;
 use object_store::memory::InMemory;
 use object_store::path::Path;
 use object_store::prefix::PrefixStore;
-use object_store::{Attribute, Attributes, ClientOptions, ObjectStore, Result};
+use object_store::{Attribute, Attributes, ClientOptions, ObjectStore, PutPayload, Result};
 use secrecy::{ExposeSecret, SecretString};
 use std::fs;
+use std::io::Cursor;
 use std::path::PathBuf;
 use std::sync::Arc;
 use tokio::fs::File;
@@ -203,6 +204,11 @@ impl Storage {
         apply_cdn_prefix(&self.cdn_prefix, &readme_path(name, version)).replace('+', "%2B")
     }
 
+    /// Returns the URL of an uploaded RSS feed.
+    pub fn feed_url(&self, feed_id: &FeedId) -> String {
+        apply_cdn_prefix(&self.cdn_prefix, &feed_id.into()).replace('+', "%2B")
+    }
+
     #[instrument(skip(self))]
     pub async fn delete_all_crate_files(&self, name: &str) -> Result<()> {
         let prefix = format!("{PREFIX_CRATES}/{name}").into();
@@ -251,6 +257,25 @@ impl Storage {
         Ok(())
     }
 
+    #[instrument(skip(self, channel))]
+    pub async fn upload_feed(
+        &self,
+        feed_id: &FeedId,
+        channel: &rss::Channel,
+    ) -> anyhow::Result<()> {
+        let path = feed_id.into();
+
+        let mut buffer = Vec::new();
+        let mut cursor = Cursor::new(&mut buffer);
+        channel.pretty_write_to(&mut cursor, b' ', 4)?;
+        let payload = PutPayload::from_bytes(buffer.into());
+
+        let attributes = self.attrs([(Attribute::ContentType, "text/xml; charset=UTF-8")]);
+        let opts = attributes.into();
+        self.store.put_opts(&path, payload, opts).await?;
+        Ok(())
+    }
+
     #[instrument(skip(self, content))]
     pub async fn sync_index(&self, name: &str, content: Option<String>) -> Result<()> {
         let path = crates_io_index::Repository::relative_index_file_for_url(name).into();
@@ -349,6 +374,19 @@ fn apply_cdn_prefix(cdn_prefix: &Option<String>, path: &Path) -> String {
     }
 }
 
+#[derive(Debug)]
+pub enum FeedId {
+    Updates,
+}
+
+impl From<&FeedId> for Path {
+    fn from(feed_id: &FeedId) -> Path {
+        match feed_id {
+            FeedId::Updates => "rss/updates.xml".into(),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/storage.rs

@@ -12,6 +12,7 @@ pub mod dump_db;
 mod expiry_notification;
 mod git;
 mod readmes;
+pub mod rss;
 mod sync_admins;
 mod typosquat;
 mod update_default_version;

src/worker/jobs/mod.rs

@@ -0,0 +1,3 @@
+mod sync_updates_feed;
+
+pub use sync_updates_feed::SyncUpdatesFeed;

src/worker/jobs/rss/mod.rs

@@ -0,0 +1,105 @@
+use crate::schema::{crates, versions};
+use crate::storage::FeedId;
+use crate::worker::Environment;
+use anyhow::anyhow;
+use crates_io_worker::BackgroundJob;
+use diesel::prelude::*;
+use std::sync::Arc;
+
+#[derive(Serialize, Deserialize)]
+pub struct SyncUpdatesFeed;
+
+const NUM_ITEMS: i64 = 100;
+
+impl BackgroundJob for SyncUpdatesFeed {
+    const JOB_NAME: &'static str = "sync_updates_feed";
+
+    type Context = Arc<Environment>;
+
+    async fn run(&self, ctx: Self::Context) -> anyhow::Result<()> {
+        let feed_id = FeedId::Updates;
+
+        info!("Loading latest {NUM_ITEMS} version updates from the database…");
+        let conn = ctx.deadpool.get().await?;
+        let version_updates = conn
+            .interact(load_version_updates)
+            .await
+            .map_err(|err| anyhow!(err.to_string()))??;
+
+        let link = rss::extension::atom::Link {
+            href: ctx.storage.feed_url(&feed_id),
+            rel: "self".to_string(),
+            mime_type: Some("application/rss+xml".to_string()),
+            ..Default::default()
+        };
+
+        let channel = rss::Channel {
+            title: "crates.io recent updates".to_string(),
+            link: "https://crates.io/".to_string(),
+            description: "Recent version publishes on the crates.io package repository".to_string(),
+            language: Some("en".to_string()),
+            atom_ext: Some(rss::extension::atom::AtomExtension { links: vec![link] }),
+            items: version_updates.into_iter().map(Into::into).collect(),
+            ..Default::default()
+        };
+
+        info!("Uploading feed to storage…");
+        ctx.storage.upload_feed(&feed_id, &channel).await?;
+
+        if let Some(cloudfront) = ctx.cloudfront() {
+            let path = object_store::path::Path::from(&feed_id);
+
+            info!(%path, "Invalidating CloudFront cache…");
+            cloudfront.invalidate(path.as_ref()).await?;
+        } else {
+            info!("Skipping CloudFront cache invalidation (CloudFront not configured)");
+        }
+
+        info!("Finished syncing updates feed");
+        Ok(())
+    }
+}
+
+fn load_version_updates(conn: &mut PgConnection) -> QueryResult<Vec<VersionUpdate>> {
+    versions::table
+        .inner_join(crates::table)
+        .order(versions::created_at.desc())
+        .select(VersionUpdate::as_select())
+        .limit(NUM_ITEMS)
+        .load(conn)
+}
+
+#[derive(Debug, Queryable, Selectable)]
+#[diesel(check_for_backend(diesel::pg::Pg))]
+struct VersionUpdate {
+    #[diesel(select_expression = crates::columns::name)]
+    name: String,
+    #[diesel(select_expression = versions::columns::num)]
+    version: String,
+    #[diesel(select_expression = crates::columns::description)]
+    description: Option<String>,
+    #[diesel(select_expression = versions::columns::created_at)]
+    time: chrono::NaiveDateTime,
+}
+
+impl From<VersionUpdate> for rss::Item {
+    fn from(u: VersionUpdate) -> Self {
+        let title = format!("{} v{}", u.name, u.version);
+        let link = format!("https://crates.io/crates/{}/{}", u.name, u.version);
+        let pub_date = u.time.and_utc().to_rfc2822();
+
+        let guid = rss::Guid {
+            value: link.clone(),
+            permalink: true,
+        };
+
+        rss::Item {
+            guid: Some(guid),
+            title: Some(title),
+            link: Some(link),
+            description: u.description,
+            pub_date: Some(pub_date),
+            ..Default::default()
+        }
+    }
+}

src/worker/jobs/rss/sync_updates_feed.rs

@@ -35,5 +35,6 @@ impl RunnerExt for Runner<Arc<Environment>> {
             .register_job_type::<jobs::UpdateDownloads>()
             .register_job_type::<jobs::UpdateDefaultVersion>()
             .register_job_type::<jobs::SendTokenExpiryNotifications>()
+            .register_job_type::<jobs::rss::SyncUpdatesFeed>()
     }
 }