From 4f80b5e9a84f9a1a60b5b51f22db6740e3fee954 Mon Sep 17 00:00:00 2001
From: Tobias Bieniek <tobias@bieniek.cloud>
Date: Fri, 4 Oct 2024 11:40:02 +0200
Subject: [PATCH 1/6] worker/jobs: Rename `git` module to `index`

This also contains the `SyncToSparseIndex` job, which does not use git at all...
---
 src/worker/jobs/{git.rs => index.rs} | 0
 src/worker/jobs/mod.rs               | 4 ++--
 2 files changed, 2 insertions(+), 2 deletions(-)
 rename src/worker/jobs/{git.rs => index.rs} (100%)

diff --git a/src/worker/jobs/git.rs b/src/worker/jobs/index.rs
similarity index 100%
rename from src/worker/jobs/git.rs
rename to src/worker/jobs/index.rs
diff --git a/src/worker/jobs/mod.rs b/src/worker/jobs/mod.rs
index 36fd5042a9..2ba1a34f81 100644
--- a/src/worker/jobs/mod.rs
+++ b/src/worker/jobs/mod.rs
@@ -12,7 +12,7 @@ mod delete_crate;
 mod downloads;
 pub mod dump_db;
 mod expiry_notification;
-mod git;
+mod index;
 mod index_version_downloads_archive;
 mod readmes;
 pub mod rss;
@@ -29,7 +29,7 @@ pub use self::downloads::{
 };
 pub use self::dump_db::DumpDb;
 pub use self::expiry_notification::SendTokenExpiryNotifications;
-pub use self::git::{NormalizeIndex, SquashIndex, SyncToGitIndex, SyncToSparseIndex};
+pub use self::index::{NormalizeIndex, SquashIndex, SyncToGitIndex, SyncToSparseIndex};
 pub use self::index_version_downloads_archive::IndexVersionDownloadsArchive;
 pub use self::readmes::RenderAndUploadReadme;
 pub use self::send_publish_notifications::SendPublishNotificationsJob;

From 6caabd9ddfeaea23508b37b6510861efbc07f6b6 Mon Sep 17 00:00:00 2001
From: Tobias Bieniek <tobias@bieniek.cloud>
Date: Fri, 4 Oct 2024 11:40:35 +0200
Subject: [PATCH 2/6] worker/jobs/index: Convert to directory module

---
 src/worker/jobs/{index.rs => index/mod.rs} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename src/worker/jobs/{index.rs => index/mod.rs} (100%)

diff --git a/src/worker/jobs/index.rs b/src/worker/jobs/index/mod.rs
similarity index 100%
rename from src/worker/jobs/index.rs
rename to src/worker/jobs/index/mod.rs

From 21b91229bfbe87aa53f7e55aac19fa642d22ba4c Mon Sep 17 00:00:00 2001
From: Tobias Bieniek <tobias@bieniek.cloud>
Date: Fri, 4 Oct 2024 11:42:49 +0200
Subject: [PATCH 3/6] worker/jobs/index: Extract `NormalizeIndex` into
 dedicated module

---
 src/worker/jobs/index/mod.rs       | 99 ++----------------------------
 src/worker/jobs/index/normalize.rs | 99 ++++++++++++++++++++++++++++++
 2 files changed, 105 insertions(+), 93 deletions(-)
 create mode 100644 src/worker/jobs/index/normalize.rs

diff --git a/src/worker/jobs/index/mod.rs b/src/worker/jobs/index/mod.rs
index 9fe7a46d82..3d60f1800b 100644
--- a/src/worker/jobs/index/mod.rs
+++ b/src/worker/jobs/index/mod.rs
@@ -5,17 +5,21 @@ use crate::worker::Environment;
 use anyhow::Context;
 use chrono::Utc;
 use crates_io_env_vars::var_parsed;
-use crates_io_index::{Crate, Repository};
+use crates_io_index::Repository;
 use crates_io_worker::BackgroundJob;
 use diesel::prelude::*;
 use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
 use sentry::Level;
 use std::fs::{self, File};
-use std::io::{BufRead, BufReader, ErrorKind, Write};
+use std::io::{ErrorKind, Write};
 use std::process::Command;
 use std::sync::Arc;
 use url::Url;
 
+mod normalize;
+
+pub use normalize::NormalizeIndex;
+
 #[derive(Serialize, Deserialize)]
 pub struct SyncToGitIndex {
     krate: String,
@@ -221,94 +225,3 @@ impl BackgroundJob for SquashIndex {
         .await
     }
 }
-
-#[derive(Serialize, Deserialize)]
-pub struct NormalizeIndex {
-    dry_run: bool,
-}
-
-impl NormalizeIndex {
-    pub fn new(dry_run: bool) -> Self {
-        Self { dry_run }
-    }
-}
-
-impl BackgroundJob for NormalizeIndex {
-    const JOB_NAME: &'static str = "normalize_index";
-    const QUEUE: &'static str = "repository";
-
-    type Context = Arc<Environment>;
-
-    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
-        info!("Normalizing the index");
-
-        let dry_run = self.dry_run;
-        spawn_blocking(move || {
-            let repo = env.lock_index()?;
-
-            let files = repo.get_files_modified_since(None)?;
-            let num_files = files.len();
-
-            for (i, file) in files.iter().enumerate() {
-                if i % 50 == 0 {
-                    info!(num_files, i, ?file);
-                }
-
-                let crate_name = file.file_name().unwrap().to_str().unwrap();
-                let path = repo.index_file(crate_name);
-                if !path.exists() {
-                    continue;
-                }
-
-                let mut body: Vec<u8> = Vec::new();
-                let file = fs::File::open(&path)?;
-                let reader = BufReader::new(file);
-                let mut versions = Vec::new();
-                for line in reader.lines() {
-                    let line = line?;
-                    if line.is_empty() {
-                        continue;
-                    }
-
-                    let mut krate: Crate = serde_json::from_str(&line)?;
-                    for dep in &mut krate.deps {
-                        // Remove deps with empty features
-                        dep.features.retain(|d| !d.is_empty());
-                        // Set null DependencyKind to Normal
-                        dep.kind =
-                            Some(dep.kind.unwrap_or(crates_io_index::DependencyKind::Normal));
-                    }
-                    krate.deps.sort();
-                    versions.push(krate);
-                }
-                for version in versions {
-                    serde_json::to_writer(&mut body, &version).unwrap();
-                    body.push(b'\n');
-                }
-                fs::write(path, body)?;
-            }
-
-            info!("Committing normalization");
-            let msg = "Normalize index format\n\n\
-        More information can be found at https://github.com/rust-lang/crates.io/pull/5066";
-            repo.run_command(Command::new("git").args(["commit", "-am", msg]))?;
-
-            let branch = match dry_run {
-                false => "master",
-                true => "normalization-dry-run",
-            };
-
-            info!(?branch, "Pushing to upstream repository");
-            repo.run_command(Command::new("git").args([
-                "push",
-                "origin",
-                &format!("HEAD:{branch}"),
-            ]))?;
-
-            info!("Index normalization completed");
-
-            Ok(())
-        })
-        .await
-    }
-}
diff --git a/src/worker/jobs/index/normalize.rs b/src/worker/jobs/index/normalize.rs
new file mode 100644
index 0000000000..7f9cba8c9a
--- /dev/null
+++ b/src/worker/jobs/index/normalize.rs
@@ -0,0 +1,99 @@
+use crate::tasks::spawn_blocking;
+use crate::worker::Environment;
+use crates_io_index::Crate;
+use crates_io_worker::BackgroundJob;
+use std::fs;
+use std::io::{BufRead, BufReader};
+use std::process::Command;
+use std::sync::Arc;
+
+#[derive(Serialize, Deserialize)]
+pub struct NormalizeIndex {
+    dry_run: bool,
+}
+
+impl NormalizeIndex {
+    pub fn new(dry_run: bool) -> Self {
+        Self { dry_run }
+    }
+}
+
+impl BackgroundJob for NormalizeIndex {
+    const JOB_NAME: &'static str = "normalize_index";
+    const QUEUE: &'static str = "repository";
+
+    type Context = Arc<Environment>;
+
+    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
+        info!("Normalizing the index");
+
+        let dry_run = self.dry_run;
+        spawn_blocking(move || {
+            let repo = env.lock_index()?;
+
+            let files = repo.get_files_modified_since(None)?;
+            let num_files = files.len();
+
+            for (i, file) in files.iter().enumerate() {
+                if i % 50 == 0 {
+                    info!(num_files, i, ?file);
+                }
+
+                let crate_name = file.file_name().unwrap().to_str().unwrap();
+                let path = repo.index_file(crate_name);
+                if !path.exists() {
+                    continue;
+                }
+
+                let mut body: Vec<u8> = Vec::new();
+                let file = fs::File::open(&path)?;
+                let reader = BufReader::new(file);
+                let mut versions = Vec::new();
+                for line in reader.lines() {
+                    let line = line?;
+                    if line.is_empty() {
+                        continue;
+                    }
+
+                    let mut krate: Crate = serde_json::from_str(&line)?;
+                    for dep in &mut krate.deps {
+                        // Remove deps with empty features
+                        dep.features.retain(|d| !d.is_empty());
+                        // Set null DependencyKind to Normal
+                        dep.kind =
+                            Some(dep.kind.unwrap_or(crates_io_index::DependencyKind::Normal));
+                    }
+                    krate.deps.sort();
+                    versions.push(krate);
+                }
+                for version in versions {
+                    serde_json::to_writer(&mut body, &version).unwrap();
+                    body.push(b'\n');
+                }
+                fs::write(path, body)?;
+            }
+
+            info!("Committing normalization");
+            let msg = "Normalize index format\n\n\
+        More information can be found at https://github.com/rust-lang/crates.io/pull/5066";
+            repo.run_command(Command::new("git").args(["commit", "-am", msg]))?;
+
+            let branch = match dry_run {
+                false => "master",
+                true => "normalization-dry-run",
+            };
+
+            info!(?branch, "Pushing to upstream repository");
+            repo.run_command(Command::new("git").args([
+                "push",
+                "origin",
+                &format!("HEAD:{branch}"),
+            ]))?;
+
+            info!("Index normalization completed");
+
+            Ok(())
+        })
+        .await
+    }
+}

From 77befd4272f3d154c3492dc17067d02debdaa570 Mon Sep 17 00:00:00 2001
From: Tobias Bieniek <tobias@bieniek.cloud>
Date: Fri, 4 Oct 2024 11:43:51 +0200
Subject: [PATCH 4/6] worker/jobs/index: Extract `SquashIndex` into dedicated
 module

---
 src/worker/jobs/index/mod.rs    | 64 +-------------------------------
 src/worker/jobs/index/squash.rs | 66 +++++++++++++++++++++++++++++++++
 2 files changed, 68 insertions(+), 62 deletions(-)
 create mode 100644 src/worker/jobs/index/squash.rs

diff --git a/src/worker/jobs/index/mod.rs b/src/worker/jobs/index/mod.rs
index 3d60f1800b..bb6322d1f8 100644
--- a/src/worker/jobs/index/mod.rs
+++ b/src/worker/jobs/index/mod.rs
@@ -3,8 +3,6 @@ use crate::tasks::spawn_blocking;
 use crate::util::diesel::Conn;
 use crate::worker::Environment;
 use anyhow::Context;
-use chrono::Utc;
-use crates_io_env_vars::var_parsed;
 use crates_io_index::Repository;
 use crates_io_worker::BackgroundJob;
 use diesel::prelude::*;
@@ -12,13 +10,13 @@ use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
 use sentry::Level;
 use std::fs::{self, File};
 use std::io::{ErrorKind, Write};
-use std::process::Command;
 use std::sync::Arc;
-use url::Url;
 
 mod normalize;
+mod squash;
 
 pub use normalize::NormalizeIndex;
+pub use squash::SquashIndex;
 
 #[derive(Serialize, Deserialize)]
 pub struct SyncToGitIndex {
@@ -167,61 +165,3 @@ pub fn get_index_data(name: &str, conn: &mut impl Conn) -> anyhow::Result<Option
 
     Ok(Some(str))
 }
-
-#[derive(Serialize, Deserialize)]
-pub struct SquashIndex;
-
-impl BackgroundJob for SquashIndex {
-    const JOB_NAME: &'static str = "squash_index";
-    const QUEUE: &'static str = "repository";
-
-    type Context = Arc<Environment>;
-
-    /// Collapse the index into a single commit, archiving the current history in a snapshot branch.
-    #[instrument(skip_all)]
-    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
-        info!("Squashing the index into a single commit");
-
-        spawn_blocking(move || {
-            let repo = env.lock_index()?;
-
-            let now = Utc::now().format("%Y-%m-%d");
-            let original_head = repo.head_oid()?.to_string();
-            let msg = format!("Collapse index into one commit\n\n\
-            Previous HEAD was {original_head}, now on the `snapshot-{now}` branch\n\n\
-            More information about this change can be found [online] and on [this issue].\n\n\
-            [online]: https://internals.rust-lang.org/t/cargos-crate-index-upcoming-squash-into-one-commit/8440\n\
-            [this issue]: https://github.com/rust-lang/crates-io-cargo-teams/issues/47");
-
-            repo.squash_to_single_commit(&msg)?;
-
-            // Shell out to git because libgit2 does not currently support push leases
-
-            repo.run_command(Command::new("git").args([
-                "push",
-                // Both updates should succeed or fail together
-                "--atomic",
-                "origin",
-                // Overwrite master, but only if it server matches the expected value
-                &format!("--force-with-lease=refs/heads/master:{original_head}"),
-                // The new squashed commit is pushed to master
-                "HEAD:refs/heads/master",
-                // The previous value of HEAD is pushed to a snapshot branch
-                &format!("{original_head}:refs/heads/snapshot-{now}"),
-            ]))?;
-
-            if let Some(archive_url) = var_parsed::<Url>("GIT_ARCHIVE_REPO_URL")? {
-                repo.run_command(Command::new("git").args([
-                    "push",
-                    archive_url.as_str(),
-                    &format!("{original_head}:snapshot-{now}"),
-                ]))?;
-            }
-
-            info!("The index has been successfully squashed.");
-
-            Ok(())
-        })
-        .await
-    }
-}
diff --git a/src/worker/jobs/index/squash.rs b/src/worker/jobs/index/squash.rs
new file mode 100644
index 0000000000..6ed52e605b
--- /dev/null
+++ b/src/worker/jobs/index/squash.rs
@@ -0,0 +1,66 @@
+use crate::tasks::spawn_blocking;
+use crate::worker::Environment;
+use chrono::Utc;
+use crates_io_env_vars::var_parsed;
+use crates_io_worker::BackgroundJob;
+use std::process::Command;
+use std::sync::Arc;
+use url::Url;
+
+#[derive(Serialize, Deserialize)]
+pub struct SquashIndex;
+
+impl BackgroundJob for SquashIndex {
+    const JOB_NAME: &'static str = "squash_index";
+    const QUEUE: &'static str = "repository";
+
+    type Context = Arc<Environment>;
+
+    /// Collapse the index into a single commit, archiving the current history in a snapshot branch.
+    #[instrument(skip_all)]
+    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
+        info!("Squashing the index into a single commit");
+
+        spawn_blocking(move || {
+            let repo = env.lock_index()?;
+
+            let now = Utc::now().format("%Y-%m-%d");
+            let original_head = repo.head_oid()?.to_string();
+            let msg = format!("Collapse index into one commit\n\n\
+            Previous HEAD was {original_head}, now on the `snapshot-{now}` branch\n\n\
+            More information about this change can be found [online] and on [this issue].\n\n\
+            [online]: https://internals.rust-lang.org/t/cargos-crate-index-upcoming-squash-into-one-commit/8440\n\
+            [this issue]: https://github.com/rust-lang/crates-io-cargo-teams/issues/47");
+
+            repo.squash_to_single_commit(&msg)?;
+
+            // Shell out to git because libgit2 does not currently support push leases
+
+            repo.run_command(Command::new("git").args([
+                "push",
+                // Both updates should succeed or fail together
+                "--atomic",
+                "origin",
+                // Overwrite master, but only if it server matches the expected value
+                &format!("--force-with-lease=refs/heads/master:{original_head}"),
+                // The new squashed commit is pushed to master
+                "HEAD:refs/heads/master",
+                // The previous value of HEAD is pushed to a snapshot branch
+                &format!("{original_head}:refs/heads/snapshot-{now}"),
+            ]))?;
+
+            if let Some(archive_url) = var_parsed::<Url>("GIT_ARCHIVE_REPO_URL")? {
+                repo.run_command(Command::new("git").args([
+                    "push",
+                    archive_url.as_str(),
+                    &format!("{original_head}:snapshot-{now}"),
+                ]))?;
+            }
+
+            info!("The index has been successfully squashed.");
+
+            Ok(())
+        })
+        .await
+    }
+}

From 9f9a94a5d446d3a68f271dce467200da1c66f1da Mon Sep 17 00:00:00 2001
From: Tobias Bieniek <tobias@bieniek.cloud>
Date: Fri, 4 Oct 2024 11:46:18 +0200
Subject: [PATCH 5/6] worker/jobs/index: Extract sync jobs into dedicated
 module

---
 src/worker/jobs/index/mod.rs  | 164 +---------------------------------
 src/worker/jobs/index/sync.rs | 162 +++++++++++++++++++++++++++++++++
 2 files changed, 164 insertions(+), 162 deletions(-)
 create mode 100644 src/worker/jobs/index/sync.rs

diff --git a/src/worker/jobs/index/mod.rs b/src/worker/jobs/index/mod.rs
index bb6322d1f8..999925c75b 100644
--- a/src/worker/jobs/index/mod.rs
+++ b/src/worker/jobs/index/mod.rs
@@ -1,167 +1,7 @@
-use crate::models;
-use crate::tasks::spawn_blocking;
-use crate::util::diesel::Conn;
-use crate::worker::Environment;
-use anyhow::Context;
-use crates_io_index::Repository;
-use crates_io_worker::BackgroundJob;
-use diesel::prelude::*;
-use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
-use sentry::Level;
-use std::fs::{self, File};
-use std::io::{ErrorKind, Write};
-use std::sync::Arc;
-
 mod normalize;
 mod squash;
+mod sync;
 
 pub use normalize::NormalizeIndex;
 pub use squash::SquashIndex;
-
-#[derive(Serialize, Deserialize)]
-pub struct SyncToGitIndex {
-    krate: String,
-}
-
-impl SyncToGitIndex {
-    pub fn new(krate: impl Into<String>) -> Self {
-        let krate = krate.into();
-        Self { krate }
-    }
-}
-
-impl BackgroundJob for SyncToGitIndex {
-    const JOB_NAME: &'static str = "sync_to_git_index";
-    const PRIORITY: i16 = 100;
-    const QUEUE: &'static str = "repository";
-
-    type Context = Arc<Environment>;
-
-    /// Regenerates or removes an index file for a single crate
-    #[instrument(skip_all, fields(krate.name = ? self.krate))]
-    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
-        info!("Syncing to git index");
-
-        let crate_name = self.krate.clone();
-        let conn = env.deadpool.get().await?;
-        spawn_blocking(move || {
-            let conn: &mut AsyncConnectionWrapper<_> = &mut conn.into();
-
-            let new = get_index_data(&crate_name, conn).context("Failed to get index data")?;
-
-            let repo = env.lock_index()?;
-            let dst = repo.index_file(&crate_name);
-
-            // Read the previous crate contents
-            let old = match fs::read_to_string(&dst) {
-                Ok(content) => Some(content),
-                Err(error) if error.kind() == ErrorKind::NotFound => None,
-                Err(error) => return Err(error.into()),
-            };
-
-            match (old, new) {
-                (None, Some(new)) => {
-                    fs::create_dir_all(dst.parent().unwrap())?;
-                    let mut file = File::create(&dst)?;
-                    file.write_all(new.as_bytes())?;
-                    repo.commit_and_push(&format!("Create crate `{}`", &crate_name), &dst)?;
-                }
-                (Some(old), Some(new)) if old != new => {
-                    let mut file = File::create(&dst)?;
-                    file.write_all(new.as_bytes())?;
-                    repo.commit_and_push(&format!("Update crate `{}`", &crate_name), &dst)?;
-                }
-                (Some(_old), None) => {
-                    fs::remove_file(&dst)?;
-                    repo.commit_and_push(&format!("Delete crate `{}`", &crate_name), &dst)?;
-                }
-                _ => debug!("Skipping sync because index is up-to-date"),
-            }
-
-            Ok(())
-        })
-        .await
-    }
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct SyncToSparseIndex {
-    krate: String,
-}
-
-impl SyncToSparseIndex {
-    pub fn new(krate: impl Into<String>) -> Self {
-        let krate = krate.into();
-        Self { krate }
-    }
-}
-
-impl BackgroundJob for SyncToSparseIndex {
-    const JOB_NAME: &'static str = "sync_to_sparse_index";
-    const PRIORITY: i16 = 100;
-
-    type Context = Arc<Environment>;
-
-    /// Regenerates or removes an index file for a single crate
-    #[instrument(skip_all, fields(krate.name = ?self.krate))]
-    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
-        info!("Syncing to sparse index");
-
-        let crate_name = self.krate.clone();
-        let conn = env.deadpool.get().await?;
-        let content = spawn_blocking(move || {
-            let conn: &mut AsyncConnectionWrapper<_> = &mut conn.into();
-            get_index_data(&crate_name, conn)
-        })
-        .await
-        .context("Failed to get index data")?;
-
-        let future = env.storage.sync_index(&self.krate, content);
-        future.await.context("Failed to sync index data")?;
-
-        if let Some(cloudfront) = env.cloudfront() {
-            let path = Repository::relative_index_file_for_url(&self.krate);
-
-            info!(%path, "Invalidating index file on CloudFront");
-            let future = cloudfront.invalidate(&path);
-            future.await.context("Failed to invalidate CloudFront")?;
-        }
-        Ok(())
-    }
-}
-
-#[instrument(skip_all, fields(krate.name = ?name))]
-pub fn get_index_data(name: &str, conn: &mut impl Conn) -> anyhow::Result<Option<String>> {
-    debug!("Looking up crate by name");
-    let Some(krate): Option<models::Crate> =
-        models::Crate::by_exact_name(name).first(conn).optional()?
-    else {
-        return Ok(None);
-    };
-
-    debug!("Gathering remaining index data");
-    let crates = krate
-        .index_metadata(conn)
-        .context("Failed to gather index metadata")?;
-
-    // This can sometimes happen when we delete versions upon owner request
-    // but don't realize that the crate is now left with no versions at all.
-    //
-    // In this case we will delete the crate from the index and log a warning to
-    // Sentry to clean this up in the database.
-    if crates.is_empty() {
-        let message = format!("Crate `{name}` has no versions left");
-        sentry::capture_message(&message, Level::Warning);
-
-        return Ok(None);
-    }
-
-    debug!("Serializing index data");
-    let mut bytes = Vec::new();
-    crates_io_index::write_crates(&crates, &mut bytes)
-        .context("Failed to serialize index metadata")?;
-
-    let str = String::from_utf8(bytes).context("Failed to decode index metadata as utf8")?;
-
-    Ok(Some(str))
-}
+pub use sync::{SyncToGitIndex, SyncToSparseIndex};
diff --git a/src/worker/jobs/index/sync.rs b/src/worker/jobs/index/sync.rs
new file mode 100644
index 0000000000..ff7c62907a
--- /dev/null
+++ b/src/worker/jobs/index/sync.rs
@@ -0,0 +1,162 @@
+use crate::models;
+use crate::tasks::spawn_blocking;
+use crate::util::diesel::Conn;
+use crate::worker::Environment;
+use anyhow::Context;
+use crates_io_index::Repository;
+use crates_io_worker::BackgroundJob;
+use diesel::{OptionalExtension, RunQueryDsl};
+use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
+use sentry::Level;
+use std::fs;
+use std::fs::File;
+use std::io::{ErrorKind, Write};
+use std::sync::Arc;
+
+#[derive(Serialize, Deserialize)]
+pub struct SyncToGitIndex {
+    krate: String,
+}
+
+impl SyncToGitIndex {
+    pub fn new(krate: impl Into<String>) -> Self {
+        let krate = krate.into();
+        Self { krate }
+    }
+}
+
+impl BackgroundJob for SyncToGitIndex {
+    const JOB_NAME: &'static str = "sync_to_git_index";
+    const PRIORITY: i16 = 100;
+    const QUEUE: &'static str = "repository";
+
+    type Context = Arc<Environment>;
+
+    /// Regenerates or removes an index file for a single crate
+    #[instrument(skip_all, fields(krate.name = ? self.krate))]
+    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
+        info!("Syncing to git index");
+
+        let crate_name = self.krate.clone();
+        let conn = env.deadpool.get().await?;
+        spawn_blocking(move || {
+            let conn: &mut AsyncConnectionWrapper<_> = &mut conn.into();
+
+            let new = get_index_data(&crate_name, conn).context("Failed to get index data")?;
+
+            let repo = env.lock_index()?;
+            let dst = repo.index_file(&crate_name);
+
+            // Read the previous crate contents
+            let old = match fs::read_to_string(&dst) {
+                Ok(content) => Some(content),
+                Err(error) if error.kind() == ErrorKind::NotFound => None,
+                Err(error) => return Err(error.into()),
+            };
+
+            match (old, new) {
+                (None, Some(new)) => {
+                    fs::create_dir_all(dst.parent().unwrap())?;
+                    let mut file = File::create(&dst)?;
+                    file.write_all(new.as_bytes())?;
+                    repo.commit_and_push(&format!("Create crate `{}`", &crate_name), &dst)?;
+                }
+                (Some(old), Some(new)) if old != new => {
+                    let mut file = File::create(&dst)?;
+                    file.write_all(new.as_bytes())?;
+                    repo.commit_and_push(&format!("Update crate `{}`", &crate_name), &dst)?;
+                }
+                (Some(_old), None) => {
+                    fs::remove_file(&dst)?;
+                    repo.commit_and_push(&format!("Delete crate `{}`", &crate_name), &dst)?;
+                }
+                _ => debug!("Skipping sync because index is up-to-date"),
+            }
+
+            Ok(())
+        })
+        .await
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct SyncToSparseIndex {
+    krate: String,
+}
+
+impl SyncToSparseIndex {
+    pub fn new(krate: impl Into<String>) -> Self {
+        let krate = krate.into();
+        Self { krate }
+    }
+}
+
+impl BackgroundJob for SyncToSparseIndex {
+    const JOB_NAME: &'static str = "sync_to_sparse_index";
+    const PRIORITY: i16 = 100;
+
+    type Context = Arc<Environment>;
+
+    /// Regenerates or removes an index file for a single crate
+    #[instrument(skip_all, fields(krate.name = ?self.krate))]
+    async fn run(&self, env: Self::Context) -> anyhow::Result<()> {
+        info!("Syncing to sparse index");
+
+        let crate_name = self.krate.clone();
+        let conn = env.deadpool.get().await?;
+        let content = spawn_blocking(move || {
+            let conn: &mut AsyncConnectionWrapper<_> = &mut conn.into();
+            get_index_data(&crate_name, conn)
+        })
+        .await
+        .context("Failed to get index data")?;
+
+        let future = env.storage.sync_index(&self.krate, content);
+        future.await.context("Failed to sync index data")?;
+
+        if let Some(cloudfront) = env.cloudfront() {
+            let path = Repository::relative_index_file_for_url(&self.krate);
+
+            info!(%path, "Invalidating index file on CloudFront");
+            let future = cloudfront.invalidate(&path);
+            future.await.context("Failed to invalidate CloudFront")?;
+        }
+        Ok(())
+    }
+}
+
+#[instrument(skip_all, fields(krate.name = ?name))]
+fn get_index_data(name: &str, conn: &mut impl Conn) -> anyhow::Result<Option<String>> {
+    debug!("Looking up crate by name");
+    let Some(krate): Option<models::Crate> =
+        models::Crate::by_exact_name(name).first(conn).optional()?
+    else {
+        return Ok(None);
+    };
+
+    debug!("Gathering remaining index data");
+    let crates = krate
+        .index_metadata(conn)
+        .context("Failed to gather index metadata")?;
+
+    // This can sometimes happen when we delete versions upon owner request
+    // but don't realize that the crate is now left with no versions at all.
+    //
+    // In this case we will delete the crate from the index and log a warning to
+    // Sentry to clean this up in the database.
+    if crates.is_empty() {
+        let message = format!("Crate `{name}` has no versions left");
+        sentry::capture_message(&message, Level::Warning);
+
+        return Ok(None);
+    }
+
+    debug!("Serializing index data");
+    let mut bytes = Vec::new();
+    crates_io_index::write_crates(&crates, &mut bytes)
+        .context("Failed to serialize index metadata")?;
+
+    let str = String::from_utf8(bytes).context("Failed to decode index metadata as utf8")?;
+
+    Ok(Some(str))
+}

From f693720f2874706d62c0f0b7b81622db17349f58 Mon Sep 17 00:00:00 2001
From: Tobias Bieniek <tobias@bieniek.cloud>
Date: Fri, 4 Oct 2024 11:50:20 +0200
Subject: [PATCH 6/6] worker/jobs/index: Add doc comment

---
 src/worker/jobs/index/mod.rs | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/src/worker/jobs/index/mod.rs b/src/worker/jobs/index/mod.rs
index 999925c75b..2fcef38729 100644
--- a/src/worker/jobs/index/mod.rs
+++ b/src/worker/jobs/index/mod.rs
@@ -1,3 +1,6 @@
+//! This module contains all background jobs related to the git and
+//! sparse indexes.
+
 mod normalize;
 mod squash;
 mod sync;