WIP: watcher: add sqs sdk, add first draft for event payload structs

crates/bin/docs_rs_watcher/Cargo.toml

@@ -8,6 +8,7 @@ edition.workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
+aws-sdk-sqs = { version = "1.99.0", default-features = false, features = ["default-https-client", "rt-tokio"] }
 clap = { workspace = true }
 # NOTE: on the new infra, switch back from `git-https-reqwest` to `git-https` (curl) once the curl version is new enough
 crates-index = { version = "3.0.0", default-features = false, features = ["git", "git-https-reqwest", "git-performance", "parallel"] }
@@ -32,6 +33,7 @@ rayon = "1.6.1"
 sqlx = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+url = { workspace = true }
 
 [dev-dependencies]
 docs_rs_config = { path = "../../lib/docs_rs_config", features = ["testing"] }

crates/bin/docs_rs_watcher/src/config.rs

@@ -2,11 +2,14 @@ use anyhow::Result;
 use docs_rs_config::AppConfig;
 use docs_rs_env_vars::{env, maybe_env, require_env};
 use std::{path::PathBuf, time::Duration};
+use url::Url;
 
 #[derive(Debug)]
 pub struct Config {
     pub registry_index_path: PathBuf,
     pub registry_url: Option<String>,
+    pub sqs_queue_url: Option<Url>,
+    pub sqs_region: Option<String>,
 
     /// How long to wait between registry checks
     pub delay_between_registry_fetches: Duration,
@@ -29,6 +32,8 @@ impl AppConfig for Config {
         Ok(Self {
             registry_index_path: env("REGISTRY_INDEX_PATH", prefix.join("crates.io-index"))?,
             registry_url: maybe_env("REGISTRY_URL")?,
+            sqs_queue_url: maybe_env("DOCSRS_SQS_QUEUE_URL")?,
+            sqs_region: maybe_env("DOCSRS_SQS_REGION")?,
             delay_between_registry_fetches: Duration::from_secs(env::<u64>(
                 "DOCSRS_DELAY_BETWEEN_REGISTRY_FETCHES",
                 60,

crates/bin/docs_rs_watcher/src/db/delete.rs

@@ -170,14 +170,18 @@ async fn delete_version_from_database(
             format!("DELETE FROM {table} WHERE {column} IN (SELECT id FROM releases WHERE crate_id = $1 AND version = $2)").as_str())
         .bind(crate_id).bind(version).execute(&mut *transaction).await?;
     }
-    let is_library: bool = sqlx::query_scalar!(
+    let Some(is_library) = sqlx::query_scalar!(
         "DELETE FROM releases WHERE crate_id = $1 AND version = $2 RETURNING is_library",
         crate_id.0,
         version as _,
     )
-    .fetch_one(&mut *transaction)
+    .fetch_optional(&mut *transaction)
     .await?
-    .unwrap_or(false);
+    else {
+        transaction.commit().await?;
+        return Ok(false);
+    };
+    let is_library = is_library.unwrap_or(false);
 
     sqlx::query!(
         "DELETE FROM queue WHERE name = $1 AND version = $2;",
@@ -690,6 +694,32 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_delete_already_deleted_version_doesnt_error() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let mut conn = env.async_conn().await?;
+
+        env.fake_release()
+            .await
+            .name(&KRATE)
+            .version(V1)
+            .create()
+            .await?;
+        env.fake_release()
+            .await
+            .name(&KRATE)
+            .version(V2)
+            .create()
+            .await?;
+
+        delete_version(&mut conn, env.storage()?, env.config(), &KRATE, &V1).await?;
+        delete_version(&mut conn, env.storage()?, env.config(), &KRATE, &V1).await?;
+
+        assert!(crate_exists(&mut conn, &KRATE).await?);
+
+        Ok(())
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_delete_version_waits_for_locked_queue_rows() -> Result<()> {
         let env = TestEnvironment::new().await?;

crates/lib/docs_rs_crates_io/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "docs_rs_crates_io"
+version = "0.1.0"
+description = "types & logic for the direct integration between docs.rs & crates.io"
+
+authors.workspace = true
+license.workspace = true
+repository.workspace = true
+edition.workspace = true
+
+[dependencies]
+chrono = { version = "0.4", features = ["serde"] }
+serde = { version = "1", features = ["derive"] }
+semver = { version = "1", features = ["serde"] }
+
+[dev-dependencies]
+serde_json = "1.0"
+
+[lints]
+workspace = true