From 95f8cc49bc3da5c527b247d7474dd30ec2b343fd Mon Sep 17 00:00:00 2001 From: Mikola Lysenko Date: Wed, 27 May 2026 15:48:08 -0400 Subject: [PATCH 1/2] refactor: dedupe ecosystem dispatch, telemetry, API, cleanup, get/blob helpers Net ~718 lines removed across six files; all 1308 tests still pass. - ecosystem_dispatch.rs: macro-extract the 8 per-ecosystem scan blocks duplicated between find_packages_for_purls and find_packages_for_rollback; share a dispatch_find core with a pypi_merge callback for the only real divergence - commands/get.rs: hoist report_error / print_json / empty_result_json / report_fetch_failure / write_all_patch_blobs / vulnerabilities_for_manifest / build_patch_record; fold the three CVE/GHSA/PURL search arms into one dispatch - utils/telemetry.rs: replace ~18 hand-built HashMap constructions with a fire() helper plus serde_json::json!({...}) literals - api/client.rs: collapse search_patches_by_{cve,ghsa,package} behind a shared search_patches_by_route(route, identifier) - api/blob_fetcher.rs: derive Default for FetchMissingBlobsResult; share an all_failed_result helper for the three mkdir-blocker branches - utils/cleanup_blobs.rs: unify cleanup_unused_blobs / cleanup_unused_archives behind a cleanup_dir(dir, dry_run, is_used) core Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/socket-patch-cli/src/commands/get.rs | 582 ++++------- .../src/ecosystem_dispatch.rs | 971 ++++++------------ .../socket-patch-core/src/api/blob_fetcher.rs | 116 +-- crates/socket-patch-core/src/api/client.rs | 54 +- .../src/utils/cleanup_blobs.rs | 161 +-- .../socket-patch-core/src/utils/telemetry.rs | 490 ++++----- 6 files changed, 828 insertions(+), 1546 deletions(-) diff --git a/crates/socket-patch-cli/src/commands/get.rs b/crates/socket-patch-cli/src/commands/get.rs index a31e2192..46133442 100644 --- a/crates/socket-patch-cli/src/commands/get.rs +++ b/crates/socket-patch-cli/src/commands/get.rs @@ -16,7 +16,7 @@ use socket_patch_core::utils::purl::is_purl; use socket_patch_core::utils::telemetry::{track_patch_fetch_failed, track_patch_fetched}; use std::collections::HashMap; use std::fmt; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use crate::args::{apply_env_toggles, GlobalArgs}; use crate::ecosystem_dispatch::crawl_all_ecosystems; @@ -159,6 +159,143 @@ fn merge_metadata(record: &mut serde_json::Value, meta: serde_json::Value) { } } +/// Print a `serde_json::Value` as pretty JSON to stdout. +fn print_json(v: &serde_json::Value) { + println!("{}", serde_json::to_string_pretty(v).unwrap()); +} + +/// Build a no-results JSON envelope with the given status code. Used in +/// the `no_packages`, `no_match`, and `not_found` branches of `get`, +/// which all share the same `{status, counts, patches: []}` shape. +fn empty_result_json(status: &str) -> serde_json::Value { + serde_json::json!({ + "status": status, + "found": 0, + "downloaded": 0, + "applied": 0, + "patches": [], + }) +} + +/// Fire a `patch_fetch_failed` telemetry event and surface the error to +/// the caller (JSON envelope or stderr). Returns `1` so callers can +/// just `return report_fetch_failure(...).await;`. +async fn report_fetch_failure( + identifier: &str, + error: impl std::fmt::Display, + fallback_to_proxy: bool, + api_token: Option<&str>, + org_slug: Option<&str>, + json: bool, +) -> i32 { + let msg = error.to_string(); + track_patch_fetch_failed(identifier, &msg, fallback_to_proxy, api_token, org_slug).await; + report_error(json, msg); + 1 +} + +/// Report an error to the caller: a `{status, error}` envelope on +/// stdout when `json` is true, otherwise a plain `Error: ...` on stderr. +fn report_error(json: bool, message: impl std::fmt::Display) { + let message = message.to_string(); + if json { + print_json(&serde_json::json!({"status": "error", "error": message})); + } else { + eprintln!("Error: {message}"); + } +} + +/// Decode a base64 string and write it to `blobs_dir/hash`. Returns a +/// formatted error string referencing `file_path` and `label` on failure. +async fn write_blob_entry( + blobs_dir: &Path, + b64: &str, + hash: &str, + file_path: &str, + label: &str, +) -> Result<(), String> { + let decoded = base64_decode(b64) + .map_err(|e| format!("Failed to decode {label} for {file_path}: {e}"))?; + tokio::fs::write(blobs_dir.join(hash), &decoded) + .await + .map_err(|e| format!("Failed to write {label} for {file_path}: {e}")) +} + +/// Write every after/before blob for `patch` into `blobs_dir`, reporting +/// per-file failures on stderr unless `quiet` is set. Returns `Err(())` +/// on the first failure; callers handle the bookkeeping that follows. +async fn write_all_patch_blobs( + blobs_dir: &Path, + patch: &PatchResponse, + quiet: bool, +) -> Result<(), ()> { + for (file_path, file_info) in &patch.files { + if let (Some(blob), Some(hash)) = + (&file_info.blob_content, &file_info.after_hash) + { + if let Err(e) = write_blob_entry(blobs_dir, blob, hash, file_path, "blob").await { + if !quiet { + eprintln!(" [error] {e}"); + } + return Err(()); + } + } + if let (Some(blob), Some(hash)) = + (&file_info.before_blob_content, &file_info.before_hash) + { + if let Err(e) = + write_blob_entry(blobs_dir, blob, hash, file_path, "before-blob").await + { + if !quiet { + eprintln!(" [error] {e}"); + } + return Err(()); + } + } + } + Ok(()) +} + +/// Convert the API-shaped vulnerability map on `PatchResponse` into the +/// serialization-shaped map stored in the manifest. +fn vulnerabilities_for_manifest( + vulns: &HashMap, +) -> HashMap { + vulns + .iter() + .map(|(id, v)| { + ( + id.clone(), + VulnerabilityInfo { + cves: v.cves.clone(), + summary: v.summary.clone(), + severity: v.severity.clone(), + description: v.description.clone(), + }, + ) + }) + .collect() +} + +/// Build the `PatchRecord` that will be inserted into the manifest for +/// `patch`. `files` is the (purl-keyed) before/after-hash map the +/// caller built — semantics for what counts as a "patchable file" differ +/// between the get and download flows, so the caller owns that decision. +fn build_patch_record( + patch: &PatchResponse, + files: HashMap, +) -> PatchRecord { + PatchRecord { + uuid: patch.uuid.clone(), + exported_at: patch.published_at.clone(), + files, + vulnerabilities: vulnerabilities_for_manifest(&patch.vulnerabilities), + description: patch.description.clone(), + license: patch.license.clone(), + tier: patch.tier.clone(), + } +} + #[derive(Args)] pub struct GetArgs { /// Patch identifier (UUID, CVE ID, GHSA ID, PURL, or package name). @@ -192,7 +329,7 @@ pub struct GetArgs { pub one_off: bool, } -#[derive(Debug, PartialEq)] +#[derive(Debug, Clone, Copy, PartialEq)] enum IdentifierType { Uuid, Cve, @@ -394,26 +531,12 @@ pub async fn download_and_apply_patches( if let Err(e) = tokio::fs::create_dir_all(&socket_dir).await { let err = format!("Failed to create .socket directory: {}", e); - if params.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": &err, - })).unwrap()); - } else { - eprintln!("Error: {}", &err); - } + report_error(params.json, &err); return (1, serde_json::json!({"status": "error", "error": err})); } if let Err(e) = tokio::fs::create_dir_all(&blobs_dir).await { let err = format!("Failed to create blobs directory: {}", e); - if params.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": &err, - })).unwrap()); - } else { - eprintln!("Error: {}", &err); - } + report_error(params.json, &err); return (1, serde_json::json!({"status": "error", "error": err})); } @@ -469,11 +592,12 @@ pub async fn download_and_apply_patches( continue; } - // Save blob contents - let mut patch_failed = false; + // Build the manifest `files` map. Download flow requires + // BOTH before+after hash (skips new files); see + // `save_and_apply_patch` for the new-file-tolerant variant. let mut files = HashMap::new(); for (file_path, file_info) in &patch.files { - if let (Some(ref before), Some(ref after)) = + if let (Some(before), Some(after)) = (&file_info.before_hash, &file_info.after_hash) { files.insert( @@ -484,57 +608,10 @@ pub async fn download_and_apply_patches( }, ); } - - if let (Some(ref blob_content), Some(ref after_hash)) = - (&file_info.blob_content, &file_info.after_hash) - { - match base64_decode(blob_content) { - Ok(decoded) => { - let blob_path = blobs_dir.join(after_hash); - if let Err(e) = tokio::fs::write(&blob_path, &decoded).await { - if !params.json && !params.silent { - eprintln!(" [error] Failed to write blob for {}: {}", file_path, e); - } - patch_failed = true; - break; - } - } - Err(e) => { - if !params.json && !params.silent { - eprintln!(" [error] Failed to decode blob for {}: {}", file_path, e); - } - patch_failed = true; - break; - } - } - } - - // Also store beforeHash blob if present (needed for rollback) - if let (Some(ref before_blob), Some(ref before_hash)) = - (&file_info.before_blob_content, &file_info.before_hash) - { - match base64_decode(before_blob) { - Ok(decoded) => { - if let Err(e) = tokio::fs::write(blobs_dir.join(before_hash), &decoded).await { - if !params.json && !params.silent { - eprintln!(" [error] Failed to write before-blob for {}: {}", file_path, e); - } - patch_failed = true; - break; - } - } - Err(e) => { - if !params.json && !params.silent { - eprintln!(" [error] Failed to decode before-blob for {}: {}", file_path, e); - } - patch_failed = true; - break; - } - } - } } - if patch_failed { + let quiet = params.json || params.silent; + if write_all_patch_blobs(&blobs_dir, &patch, quiet).await.is_err() { patches_failed += 1; downloaded_patches.push(serde_json::json!({ "purl": patch.purl, @@ -545,34 +622,9 @@ pub async fn download_and_apply_patches( continue; } - let vulnerabilities: HashMap = patch - .vulnerabilities - .iter() - .map(|(id, v)| { - ( - id.clone(), - VulnerabilityInfo { - cves: v.cves.clone(), - summary: v.summary.clone(), - severity: v.severity.clone(), - description: v.description.clone(), - }, - ) - }) - .collect(); - - manifest.patches.insert( - patch.purl.clone(), - PatchRecord { - uuid: patch.uuid.clone(), - exported_at: patch.published_at.clone(), - files, - vulnerabilities, - description: patch.description.clone(), - license: patch.license.clone(), - tier: patch.tier.clone(), - }, - ); + manifest + .patches + .insert(patch.purl.clone(), build_patch_record(&patch, files)); let mut action_record = match &action { PatchAction::Updated { old_uuid } => { @@ -634,14 +686,12 @@ pub async fn download_and_apply_patches( // Write manifest if let Err(e) = write_manifest(&manifest_path, &manifest).await { - let err_json = serde_json::json!({ - "status": "error", - "error": format!("Error writing manifest: {e}"), - }); + let msg = format!("Error writing manifest: {e}"); + let err_json = serde_json::json!({ "status": "error", "error": &msg }); if params.json { - println!("{}", serde_json::to_string_pretty(&err_json).unwrap()); + print_json(&err_json); } else { - eprintln!("Error writing manifest: {e}"); + eprintln!("{msg}"); } return (1, err_json); } @@ -707,22 +757,18 @@ pub async fn run(args: GetArgs) -> i32 { .filter(|&&f| f) .count(); if type_flags > 1 { - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": "Only one of --id, --cve, --ghsa, or --package can be specified", - })).unwrap()); - } else { - eprintln!("Error: Only one of --id, --cve, --ghsa, or --package can be specified"); - } + report_error( + args.common.json, + "Only one of --id, --cve, --ghsa, or --package can be specified", + ); return 1; } if args.one_off && args.save_only { if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ + print_json(&serde_json::json!({ "status": "error", "error": "--one-off and --save-only cannot be used together", - })).unwrap()); + })); } else { eprintln!("Error: --one-off and --save-only cannot be used together"); } @@ -805,7 +851,7 @@ pub async fn run(args: GetArgs) -> i32 { ) .await; if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ + print_json(&serde_json::json!({ "status": "paid_required", "found": 1, "downloaded": 0, @@ -815,7 +861,7 @@ pub async fn run(args: GetArgs) -> i32 { "uuid": patch.uuid, "tier": "paid", }], - })).unwrap()); + })); } else { println!("\nThis patch requires a paid subscription to download."); println!("\n Patch: {}", patch.purl); @@ -854,129 +900,70 @@ pub async fn run(args: GetArgs) -> i32 { ) .await; if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "not_found", - "found": 0, - "downloaded": 0, - "applied": 0, - "patches": [], - })).unwrap()); + print_json(&empty_result_json("not_found")); } else { println!("No patch found with UUID: {}", args.identifier); } return 0; } Err(e) => { - track_patch_fetch_failed( + return report_fetch_failure( &args.identifier, - &e, + e, fallback_to_proxy, telemetry_token.as_deref(), telemetry_org.as_deref(), + args.common.json, ) .await; - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": e.to_string(), - })).unwrap()); - } else { - eprintln!("Error: {e}"); - } - return 1; } } } - // For CVE/GHSA/PURL/package, search first + // For CVE/GHSA/PURL/package, search first. + // CVE / GHSA / PURL share the same path: log the search, dispatch to + // the matching endpoint, and surface errors via `report_fetch_failure`. let search_response: SearchResponse = match id_type { - IdentifierType::Cve => { + IdentifierType::Cve | IdentifierType::Ghsa | IdentifierType::Purl => { if !args.common.json { - println!("Searching patches for CVE: {}", args.identifier); + let label = match id_type { + IdentifierType::Cve => "CVE", + IdentifierType::Ghsa => "GHSA", + IdentifierType::Purl => "PURL", + _ => unreachable!(), + }; + println!("Searching patches for {label}: {}", args.identifier); } - match api_client - .search_patches_by_cve(effective_org_slug, &args.identifier) - .await - { - Ok(r) => r, - Err(e) => { - track_patch_fetch_failed( - &args.identifier, - &e, - fallback_to_proxy, - telemetry_token.as_deref(), - telemetry_org.as_deref(), - ) - .await; - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": e.to_string(), - })).unwrap()); - } else { - eprintln!("Error: {e}"); - } - return 1; + let result = match id_type { + IdentifierType::Cve => { + api_client + .search_patches_by_cve(effective_org_slug, &args.identifier) + .await } - } - } - IdentifierType::Ghsa => { - if !args.common.json { - println!("Searching patches for GHSA: {}", args.identifier); - } - match api_client - .search_patches_by_ghsa(effective_org_slug, &args.identifier) - .await - { - Ok(r) => r, - Err(e) => { - track_patch_fetch_failed( - &args.identifier, - &e, - fallback_to_proxy, - telemetry_token.as_deref(), - telemetry_org.as_deref(), - ) - .await; - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": e.to_string(), - })).unwrap()); - } else { - eprintln!("Error: {e}"); - } - return 1; + IdentifierType::Ghsa => { + api_client + .search_patches_by_ghsa(effective_org_slug, &args.identifier) + .await } - } - } - IdentifierType::Purl => { - if !args.common.json { - println!("Searching patches for PURL: {}", args.identifier); - } - match api_client - .search_patches_by_package(effective_org_slug, &args.identifier) - .await - { + IdentifierType::Purl => { + api_client + .search_patches_by_package(effective_org_slug, &args.identifier) + .await + } + _ => unreachable!(), + }; + match result { Ok(r) => r, Err(e) => { - track_patch_fetch_failed( + return report_fetch_failure( &args.identifier, - &e, + e, fallback_to_proxy, telemetry_token.as_deref(), telemetry_org.as_deref(), + args.common.json, ) .await; - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": e.to_string(), - })).unwrap()); - } else { - eprintln!("Error: {e}"); - } - return 1; } } } @@ -994,13 +981,7 @@ pub async fn run(args: GetArgs) -> i32 { if all_packages.is_empty() { if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "no_packages", - "found": 0, - "downloaded": 0, - "applied": 0, - "patches": [], - })).unwrap()); + print_json(&empty_result_json("no_packages")); } else if args.common.global { println!("No global packages found."); } else { @@ -1027,13 +1008,7 @@ pub async fn run(args: GetArgs) -> i32 { if matches.is_empty() { if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "no_match", - "found": 0, - "downloaded": 0, - "applied": 0, - "patches": [], - })).unwrap()); + print_json(&empty_result_json("no_match")); } else { println!("No packages matching \"{}\" found.", args.identifier); } @@ -1055,23 +1030,15 @@ pub async fn run(args: GetArgs) -> i32 { { Ok(r) => r, Err(e) => { - track_patch_fetch_failed( + return report_fetch_failure( &args.identifier, - &e, + e, fallback_to_proxy, telemetry_token.as_deref(), telemetry_org.as_deref(), + args.common.json, ) .await; - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": e.to_string(), - })).unwrap()); - } else { - eprintln!("Error: {e}"); - } - return 1; } } } @@ -1080,13 +1047,7 @@ pub async fn run(args: GetArgs) -> i32 { if search_response.patches.is_empty() { if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "not_found", - "found": 0, - "downloaded": 0, - "applied": 0, - "patches": [], - })).unwrap()); + print_json(&empty_result_json("not_found")); } else { println!( "No patches found for {}: {}", @@ -1110,7 +1071,7 @@ pub async fn run(args: GetArgs) -> i32 { if accessible.is_empty() { if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ + print_json(&serde_json::json!({ "status": "paid_required", "found": search_response.patches.len(), "downloaded": 0, @@ -1120,7 +1081,7 @@ pub async fn run(args: GetArgs) -> i32 { "uuid": p.uuid, "tier": p.tier, })).collect::>(), - })).unwrap()); + })); } else { println!("\nAll available patches require a paid subscription."); println!("\n Upgrade at: https://socket.dev/pricing\n"); @@ -1238,27 +1199,14 @@ async fn save_and_apply_patch( Ok(Some(p)) => p, Ok(None) => { if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "not_found", - "found": 0, - "downloaded": 0, - "applied": 0, - "patches": [], - })).unwrap()); + print_json(&empty_result_json("not_found")); } else { println!("No patch found with UUID: {uuid}"); } return 0; } Err(e) => { - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": e.to_string(), - })).unwrap()); - } else { - eprintln!("Error: {e}"); - } + report_error(args.common.json, e); return 1; } }; @@ -1268,14 +1216,7 @@ async fn save_and_apply_patch( let manifest_path = socket_dir.join("manifest.json"); if let Err(e) = tokio::fs::create_dir_all(&blobs_dir).await { - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": format!("Failed to create blobs directory: {}", e), - })).unwrap()); - } else { - eprintln!("Error: Failed to create blobs directory: {}", e); - } + report_error(args.common.json, format!("Failed to create blobs directory: {e}")); return 1; } @@ -1284,72 +1225,29 @@ async fn save_and_apply_patch( _ => PatchManifest::new(), }; - // Build and save patch record - let mut blob_failed = false; + // Build the manifest `files` map. UUID flow is more permissive than + // the download flow: a file with after_hash but no before_hash is a + // new file; we record an empty `before_hash` and let apply treat it + // as a new-file insert. let mut files = HashMap::new(); for (file_path, file_info) in &patch.files { - if let Some(ref after) = file_info.after_hash { + if let Some(after) = &file_info.after_hash { files.insert( file_path.clone(), PatchFileInfo { - before_hash: file_info - .before_hash - .clone() - .unwrap_or_default(), + before_hash: file_info.before_hash.clone().unwrap_or_default(), after_hash: after.clone(), }, ); } - if let (Some(ref blob_content), Some(ref after_hash)) = - (&file_info.blob_content, &file_info.after_hash) - { - match base64_decode(blob_content) { - Ok(decoded) => { - if let Err(e) = tokio::fs::write(blobs_dir.join(after_hash), &decoded).await { - if !args.common.json { - eprintln!(" [error] Failed to write blob for {}: {}", file_path, e); - } - blob_failed = true; - break; - } - } - Err(e) => { - if !args.common.json { - eprintln!(" [error] Failed to decode blob for {}: {}", file_path, e); - } - blob_failed = true; - break; - } - } - } - // Also store beforeHash blob if present (needed for rollback) - if let (Some(ref before_blob), Some(ref before_hash)) = - (&file_info.before_blob_content, &file_info.before_hash) - { - match base64_decode(before_blob) { - Ok(decoded) => { - if let Err(e) = tokio::fs::write(blobs_dir.join(before_hash), &decoded).await { - if !args.common.json { - eprintln!(" [error] Failed to write before-blob for {}: {}", file_path, e); - } - blob_failed = true; - break; - } - } - Err(e) => { - if !args.common.json { - eprintln!(" [error] Failed to decode before-blob for {}: {}", file_path, e); - } - blob_failed = true; - break; - } - } - } } - if blob_failed { + if write_all_patch_blobs(&blobs_dir, &patch, args.common.json) + .await + .is_err() + { if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ + print_json(&serde_json::json!({ "status": "error", "found": 1, "downloaded": 0, @@ -1361,56 +1259,24 @@ async fn save_and_apply_patch( "action": "failed", "error": "Blob decode or write failed", }], - })).unwrap()); + })); } else { eprintln!("Error: Blob decode or write failed for patch {}", patch.purl); } return 1; } - let vulnerabilities: HashMap = patch - .vulnerabilities - .iter() - .map(|(id, v)| { - ( - id.clone(), - VulnerabilityInfo { - cves: v.cves.clone(), - summary: v.summary.clone(), - severity: v.severity.clone(), - description: v.description.clone(), - }, - ) - }) - .collect(); - let added = manifest .patches .get(&patch.purl) .is_none_or(|p| p.uuid != patch.uuid); - manifest.patches.insert( - patch.purl.clone(), - PatchRecord { - uuid: patch.uuid.clone(), - exported_at: patch.published_at.clone(), - files, - vulnerabilities, - description: patch.description.clone(), - license: patch.license.clone(), - tier: patch.tier.clone(), - }, - ); + manifest + .patches + .insert(patch.purl.clone(), build_patch_record(&patch, files)); if let Err(e) = write_manifest(&manifest_path, &manifest).await { - if args.common.json { - println!("{}", serde_json::to_string_pretty(&serde_json::json!({ - "status": "error", - "error": format!("Error writing manifest: {e}"), - })).unwrap()); - } else { - eprintln!("Error writing manifest: {e}"); - } + report_error(args.common.json, format!("Error writing manifest: {e}")); return 1; } diff --git a/crates/socket-patch-cli/src/ecosystem_dispatch.rs b/crates/socket-patch-cli/src/ecosystem_dispatch.rs index 4ae2dce7..444b8692 100644 --- a/crates/socket-patch-cli/src/ecosystem_dispatch.rs +++ b/crates/socket-patch-cli/src/ecosystem_dispatch.rs @@ -29,13 +29,9 @@ use socket_patch_core::crawlers::DenoCrawler; /// recovery — the user has to re-download the jar. #[cfg(feature = "maven")] fn maven_runtime_enabled() -> bool { - std::env::var("SOCKET_EXPERIMENTAL_MAVEN") - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) - .unwrap_or(false) + env_truthy("SOCKET_EXPERIMENTAL_MAVEN") } -/// One-line stderr warning for the "Maven patches present, but -/// experimental gate is off" path. #[cfg(feature = "maven")] fn warn_maven_disabled(skipped: usize) { eprintln!( @@ -46,24 +42,18 @@ fn warn_maven_disabled(skipped: usize) { eprintln!(" Set SOCKET_EXPERIMENTAL_MAVEN=1 to enable at your own risk."); } -/// Runtime opt-in gate for experimental NuGet support. -/// -/// Same shape as the Maven gate. Even with the sidecar fixup -/// deleting `.nupkg.metadata`, signed packages still carry a -/// `.nupkg.sha512` marker that NuGet treats as tamper-evidence -/// at restore time. The fixup cannot honestly rewrite this -/// without the original `.nupkg` (which we don't have post- -/// extraction). Refuse to dispatch unless the operator has -/// explicitly opted in to the experimental tier. +/// Runtime opt-in gate for experimental NuGet support. Same shape as +/// the Maven gate. Even with the sidecar fixup deleting +/// `.nupkg.metadata`, signed packages still carry a `.nupkg.sha512` +/// marker that NuGet treats as tamper-evidence at restore time. The +/// fixup cannot honestly rewrite this without the original `.nupkg` +/// (which we don't have post-extraction). Refuse to dispatch unless +/// the operator has explicitly opted in to the experimental tier. #[cfg(feature = "nuget")] fn nuget_runtime_enabled() -> bool { - std::env::var("SOCKET_EXPERIMENTAL_NUGET") - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) - .unwrap_or(false) + env_truthy("SOCKET_EXPERIMENTAL_NUGET") } -/// One-line stderr warning for the "NuGet patches present, but -/// experimental gate is off" path. #[cfg(feature = "nuget")] fn warn_nuget_disabled(skipped: usize) { eprintln!( @@ -75,13 +65,19 @@ fn warn_nuget_disabled(skipped: usize) { eprintln!(" Set SOCKET_EXPERIMENTAL_NUGET=1 to enable at your own risk."); } +#[cfg(any(feature = "maven", feature = "nuget"))] +fn env_truthy(name: &str) -> bool { + std::env::var(name) + .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) + .unwrap_or(false) +} + /// Partition PURLs by ecosystem, filtering by the `--ecosystems` flag if set. pub fn partition_purls( purls: &[String], allowed_ecosystems: Option<&[String]>, ) -> HashMap> { let mut map: HashMap> = HashMap::new(); - for purl in purls { if let Some(eco) = Ecosystem::from_purl(purl) { if let Some(allowed) = allowed_ecosystems { @@ -92,352 +88,305 @@ pub fn partition_purls( map.entry(eco).or_default().push(purl.clone()); } } - map } -/// For each ecosystem in the partitioned map, create the crawler, discover -/// source paths, and look up the given PURLs. Returns a unified -/// `purl -> path` map. -pub async fn find_packages_for_purls( - partitioned: &HashMap>, - options: &CrawlerOptions, - silent: bool, -) -> HashMap { - let mut all_packages: HashMap = HashMap::new(); - - // npm - if let Some(npm_purls) = partitioned.get(&Ecosystem::Npm) { - if !npm_purls.is_empty() { - let npm_crawler = NpmCrawler; - match npm_crawler.get_node_modules_paths(options).await { - Ok(nm_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = nm_paths.first() { - println!("Using global npm packages at: {}", first.display()); - } - } - for nm_path in &nm_paths { - match npm_crawler.find_by_purls(nm_path, npm_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", nm_path.display(), e); +/// Standard scan-one-ecosystem pattern: discover source paths, run +/// `find_by_purls` on each, and merge results into `$out` keyed by PURL +/// (first wins). Used by every ecosystem except pypi (which dedups +/// PURLs and, on rollback, remaps base PURLs back to qualified ones). +/// +/// `$using_label` is the noun in "Using at: " for global +/// scans; pass `""` to suppress that line. +macro_rules! scan_ecosystem { + ( + out = $out:ident, + partitioned = $partitioned:expr, + eco = $eco:expr, + options = $options:expr, + silent = $silent:expr, + crawler = $crawler:expr, + get_paths = $get_paths:ident, + using_label = $using_label:expr, + err_label = $err_label:expr, + purls_override = $purls_override:expr, + on_match = $on_match:expr $(,)? + ) => {{ + if let Some(purls) = $partitioned.get(&$eco) { + if !purls.is_empty() { + let crawler = $crawler; + let purls_to_use: Vec = $purls_override(purls); + match crawler.$get_paths($options).await { + Ok(paths) => { + let using: &str = $using_label; + if !using.is_empty() + && ($options.global || $options.global_prefix.is_some()) + && !$silent + { + if let Some(first) = paths.first() { + println!("Using {} at: {}", using, first.display()); + } + } + for path in &paths { + match crawler.find_by_purls(path, &purls_to_use).await { + Ok(packages) => { + $on_match(&mut $out, purls, packages); + } + Err(e) => { + if !$silent { + eprintln!( + "Warning: Failed to scan {}: {}", + path.display(), + e + ); + } } } } } - } - Err(e) => { - if !silent { - eprintln!("Failed to find npm packages: {e}"); + Err(e) => { + if !$silent { + eprintln!("Failed to find {}: {}", $err_label, e); + } } } } } + }}; +} + +/// Default merge: insert the crawler-returned PURL → first wins. +fn merge_first_wins( + out: &mut HashMap, + _purls: &[String], + packages: HashMap, +) { + for (purl, pkg) in packages { + out.entry(purl).or_insert(pkg.path); } +} - // pypi — deduplicate by base PURL (stripping qualifiers) - if let Some(pypi_purls) = partitioned.get(&Ecosystem::Pypi) { - if !pypi_purls.is_empty() { - let python_crawler = PythonCrawler; - let base_pypi_purls: Vec = pypi_purls - .iter() - .map(|p| strip_purl_qualifiers(p).to_string()) - .collect::>() - .into_iter() - .collect(); - - match python_crawler.get_site_packages_paths(options).await { - Ok(sp_paths) => { - for sp_path in &sp_paths { - match python_crawler.find_by_purls(sp_path, &base_pypi_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", sp_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Python packages: {e}"); - } - } +/// Pypi rollback merge: the crawler is queried with base PURLs (no +/// `?qualifiers`); fan the resulting paths back out to every qualified +/// caller-supplied PURL that strips to the same base. +fn merge_pypi_qualified( + out: &mut HashMap, + purls: &[String], + packages: HashMap, +) { + for (base_purl, pkg) in packages { + for qualified in purls { + if strip_purl_qualifiers(qualified) == base_purl + && !out.contains_key(qualified) + { + out.insert(qualified.clone(), pkg.path.clone()); } } } +} + +fn dedup_pypi_purls(purls: &[String]) -> Vec { + purls + .iter() + .map(|p| strip_purl_qualifiers(p).to_string()) + .collect::>() + .into_iter() + .collect() +} + +fn passthrough_purls(purls: &[String]) -> Vec { + purls.to_vec() +} + +/// Drive every enabled ecosystem's find-by-purls path, accumulating +/// into one `purl -> path` map. +/// +/// `pypi_merge` lets the rollback variant fan a single crawler result +/// out to every caller-supplied qualified PURL; everything else just +/// inserts the crawler-returned PURL with first-wins semantics. +async fn dispatch_find( + partitioned: &HashMap>, + options: &CrawlerOptions, + silent: bool, + pypi_merge: fn( + &mut HashMap, + &[String], + HashMap, + ), +) -> HashMap { + let mut out: HashMap = HashMap::new(); + + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Npm, + options = options, + silent = silent, + crawler = NpmCrawler, + get_paths = get_node_modules_paths, + using_label = "global npm packages", + err_label = "npm packages", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); + + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Pypi, + options = options, + silent = silent, + crawler = PythonCrawler, + get_paths = get_site_packages_paths, + using_label = "", + err_label = "Python packages", + purls_override = dedup_pypi_purls, + on_match = pypi_merge, + ); - // cargo #[cfg(feature = "cargo")] - if let Some(cargo_purls) = partitioned.get(&Ecosystem::Cargo) { - if !cargo_purls.is_empty() { - let cargo_crawler = CargoCrawler; - match cargo_crawler.get_crate_source_paths(options).await { - Ok(src_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = src_paths.first() { - println!("Using cargo crate sources at: {}", first.display()); - } - } - for src_path in &src_paths { - match cargo_crawler.find_by_purls(src_path, cargo_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", src_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Cargo crates: {e}"); - } - } - } - } - } + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Cargo, + options = options, + silent = silent, + crawler = CargoCrawler, + get_paths = get_crate_source_paths, + using_label = "cargo crate sources", + err_label = "Cargo crates", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); - // gem - if let Some(gem_purls) = partitioned.get(&Ecosystem::Gem) { - if !gem_purls.is_empty() { - let ruby_crawler = RubyCrawler; - match ruby_crawler.get_gem_paths(options).await { - Ok(gem_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = gem_paths.first() { - println!("Using ruby gem paths at: {}", first.display()); - } - } - for gem_path in &gem_paths { - match ruby_crawler.find_by_purls(gem_path, gem_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", gem_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Ruby gems: {e}"); - } - } - } - } - } + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Gem, + options = options, + silent = silent, + crawler = RubyCrawler, + get_paths = get_gem_paths, + using_label = "ruby gem paths", + err_label = "Ruby gems", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); - // golang #[cfg(feature = "golang")] - if let Some(golang_purls) = partitioned.get(&Ecosystem::Golang) { - if !golang_purls.is_empty() { - let go_crawler = GoCrawler; - match go_crawler.get_module_cache_paths(options).await { - Ok(cache_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = cache_paths.first() { - println!("Using Go module cache at: {}", first.display()); - } - } - for cache_path in &cache_paths { - match go_crawler.find_by_purls(cache_path, golang_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", cache_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Go modules: {e}"); - } - } - } - } - } + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Golang, + options = options, + silent = silent, + crawler = GoCrawler, + get_paths = get_module_cache_paths, + using_label = "Go module cache", + err_label = "Go modules", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); - // maven — experimental, double-gated. See `maven_runtime_enabled`. #[cfg(feature = "maven")] if let Some(maven_purls) = partitioned.get(&Ecosystem::Maven) { if !maven_purls.is_empty() && !maven_runtime_enabled() { if !silent { warn_maven_disabled(maven_purls.len()); } - } else if !maven_purls.is_empty() { - let maven_crawler = MavenCrawler; - match maven_crawler.get_maven_repo_paths(options).await { - Ok(repo_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = repo_paths.first() { - println!("Using Maven repository at: {}", first.display()); - } - } - for repo_path in &repo_paths { - match maven_crawler.find_by_purls(repo_path, maven_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", repo_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Maven packages: {e}"); - } - } - } + } else { + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Maven, + options = options, + silent = silent, + crawler = MavenCrawler, + get_paths = get_maven_repo_paths, + using_label = "Maven repository", + err_label = "Maven packages", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); } } - // composer #[cfg(feature = "composer")] - if let Some(composer_purls) = partitioned.get(&Ecosystem::Composer) { - if !composer_purls.is_empty() { - let composer_crawler = ComposerCrawler; - match composer_crawler.get_vendor_paths(options).await { - Ok(vendor_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = vendor_paths.first() { - println!("Using PHP vendor packages at: {}", first.display()); - } - } - for vendor_path in &vendor_paths { - match composer_crawler.find_by_purls(vendor_path, composer_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", vendor_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find PHP packages: {e}"); - } - } - } - } - } + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Composer, + options = options, + silent = silent, + crawler = ComposerCrawler, + get_paths = get_vendor_paths, + using_label = "PHP vendor packages", + err_label = "PHP packages", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); - // nuget — experimental, double-gated. See `nuget_runtime_enabled`. #[cfg(feature = "nuget")] if let Some(nuget_purls) = partitioned.get(&Ecosystem::Nuget) { if !nuget_purls.is_empty() && !nuget_runtime_enabled() { if !silent { warn_nuget_disabled(nuget_purls.len()); } - } else if !nuget_purls.is_empty() { - let nuget_crawler = NuGetCrawler; - match nuget_crawler.get_nuget_package_paths(options).await { - Ok(pkg_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = pkg_paths.first() { - println!("Using NuGet packages at: {}", first.display()); - } - } - for pkg_path in &pkg_paths { - match nuget_crawler.find_by_purls(pkg_path, nuget_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", pkg_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find NuGet packages: {e}"); - } - } - } + } else { + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Nuget, + options = options, + silent = silent, + crawler = NuGetCrawler, + get_paths = get_nuget_package_paths, + using_label = "NuGet packages", + err_label = "NuGet packages", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); } } - // deno — JSR registry packages cached under DENO_DIR/npm/jsr.io/. #[cfg(feature = "deno")] - if let Some(deno_purls) = partitioned.get(&Ecosystem::Deno) { - if !deno_purls.is_empty() { - let deno_crawler = DenoCrawler; - match deno_crawler.get_jsr_cache_paths(options).await { - Ok(cache_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = cache_paths.first() { - println!("Using Deno JSR cache at: {}", first.display()); - } - } - for cache_path in &cache_paths { - match deno_crawler.find_by_purls(cache_path, deno_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", cache_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Deno JSR packages: {e}"); - } - } - } - } - } + scan_ecosystem!( + out = out, + partitioned = partitioned, + eco = Ecosystem::Deno, + options = options, + silent = silent, + crawler = DenoCrawler, + get_paths = get_jsr_cache_paths, + using_label = "Deno JSR cache", + err_label = "Deno JSR packages", + purls_override = passthrough_purls, + on_match = merge_first_wins, + ); + + out +} + +/// For each ecosystem in the partitioned map, create the crawler, discover +/// source paths, and look up the given PURLs. Returns a unified +/// `purl -> path` map. +pub async fn find_packages_for_purls( + partitioned: &HashMap>, + options: &CrawlerOptions, + silent: bool, +) -> HashMap { + dispatch_find(partitioned, options, silent, merge_first_wins).await +} - all_packages +/// Variant of `find_packages_for_purls` for rollback, which needs to remap +/// pypi qualified PURLs (with `?artifact_id=...`) to the base PURL found +/// by the crawler. +pub async fn find_packages_for_rollback( + partitioned: &HashMap>, + options: &CrawlerOptions, + silent: bool, +) -> HashMap { + dispatch_find(partitioned, options, silent, merge_pypi_qualified).await } /// Crawl all enabled ecosystems and return all packages plus per-ecosystem counts. @@ -447,386 +396,40 @@ pub async fn crawl_all_ecosystems( let mut all_packages = Vec::new(); let mut counts: HashMap = HashMap::new(); - let npm_crawler = NpmCrawler; - let npm_packages = npm_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Npm, npm_packages.len()); - all_packages.extend(npm_packages); - - let python_crawler = PythonCrawler; - let python_packages = python_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Pypi, python_packages.len()); - all_packages.extend(python_packages); - - #[cfg(feature = "cargo")] - { - let cargo_crawler = CargoCrawler; - let cargo_packages = cargo_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Cargo, cargo_packages.len()); - all_packages.extend(cargo_packages); - } - - { - let ruby_crawler = RubyCrawler; - let gem_packages = ruby_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Gem, gem_packages.len()); - all_packages.extend(gem_packages); + macro_rules! crawl { + ($eco:expr, $crawler:expr) => {{ + let pkgs = $crawler.crawl_all(options).await; + counts.insert($eco, pkgs.len()); + all_packages.extend(pkgs); + }}; } + crawl!(Ecosystem::Npm, NpmCrawler); + crawl!(Ecosystem::Pypi, PythonCrawler); + #[cfg(feature = "cargo")] + crawl!(Ecosystem::Cargo, CargoCrawler); + crawl!(Ecosystem::Gem, RubyCrawler); #[cfg(feature = "golang")] - { - let go_crawler = GoCrawler; - let go_packages = go_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Golang, go_packages.len()); - all_packages.extend(go_packages); - } - + crawl!(Ecosystem::Golang, GoCrawler); #[cfg(feature = "maven")] if maven_runtime_enabled() { // Same runtime gate as `find_packages_for_purls` — `scan` // walks the Maven repo only when the operator has explicitly // opted into experimental support. - let maven_crawler = MavenCrawler; - let maven_packages = maven_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Maven, maven_packages.len()); - all_packages.extend(maven_packages); + crawl!(Ecosystem::Maven, MavenCrawler); } - #[cfg(feature = "composer")] - { - let composer_crawler = ComposerCrawler; - let composer_packages = composer_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Composer, composer_packages.len()); - all_packages.extend(composer_packages); - } - + crawl!(Ecosystem::Composer, ComposerCrawler); #[cfg(feature = "nuget")] if nuget_runtime_enabled() { - // Same runtime gate as `find_packages_for_purls`. - let nuget_crawler = NuGetCrawler; - let nuget_packages = nuget_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Nuget, nuget_packages.len()); - all_packages.extend(nuget_packages); + crawl!(Ecosystem::Nuget, NuGetCrawler); } - #[cfg(feature = "deno")] - { - let deno_crawler = DenoCrawler; - let deno_packages = deno_crawler.crawl_all(options).await; - counts.insert(Ecosystem::Deno, deno_packages.len()); - all_packages.extend(deno_packages); - } + crawl!(Ecosystem::Deno, DenoCrawler); (all_packages, counts) } -/// Variant of `find_packages_for_purls` for rollback, which needs to remap -/// pypi qualified PURLs (with `?artifact_id=...`) to the base PURL found -/// by the crawler. -pub async fn find_packages_for_rollback( - partitioned: &HashMap>, - options: &CrawlerOptions, - silent: bool, -) -> HashMap { - let mut all_packages: HashMap = HashMap::new(); - - // npm - if let Some(npm_purls) = partitioned.get(&Ecosystem::Npm) { - if !npm_purls.is_empty() { - let npm_crawler = NpmCrawler; - match npm_crawler.get_node_modules_paths(options).await { - Ok(nm_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = nm_paths.first() { - println!("Using global npm packages at: {}", first.display()); - } - } - for nm_path in &nm_paths { - match npm_crawler.find_by_purls(nm_path, npm_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", nm_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find npm packages: {e}"); - } - } - } - } - } - - // pypi — remap qualified PURLs to found base PURLs - if let Some(pypi_purls) = partitioned.get(&Ecosystem::Pypi) { - if !pypi_purls.is_empty() { - let python_crawler = PythonCrawler; - let base_pypi_purls: Vec = pypi_purls - .iter() - .map(|p| strip_purl_qualifiers(p).to_string()) - .collect::>() - .into_iter() - .collect(); - - if let Ok(sp_paths) = python_crawler.get_site_packages_paths(options).await { - for sp_path in &sp_paths { - match python_crawler.find_by_purls(sp_path, &base_pypi_purls).await { - Ok(packages) => { - for (base_purl, pkg) in packages { - for qualified_purl in pypi_purls { - if strip_purl_qualifiers(qualified_purl) == base_purl - && !all_packages.contains_key(qualified_purl) - { - all_packages - .insert(qualified_purl.clone(), pkg.path.clone()); - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", sp_path.display(), e); - } - } - } - } - } - } - } - - // cargo - #[cfg(feature = "cargo")] - if let Some(cargo_purls) = partitioned.get(&Ecosystem::Cargo) { - if !cargo_purls.is_empty() { - let cargo_crawler = CargoCrawler; - match cargo_crawler.get_crate_source_paths(options).await { - Ok(src_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = src_paths.first() { - println!("Using cargo crate sources at: {}", first.display()); - } - } - for src_path in &src_paths { - match cargo_crawler.find_by_purls(src_path, cargo_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", src_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Cargo crates: {e}"); - } - } - } - } - } - - // gem - if let Some(gem_purls) = partitioned.get(&Ecosystem::Gem) { - if !gem_purls.is_empty() { - let ruby_crawler = RubyCrawler; - match ruby_crawler.get_gem_paths(options).await { - Ok(gem_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = gem_paths.first() { - println!("Using ruby gem paths at: {}", first.display()); - } - } - for gem_path in &gem_paths { - match ruby_crawler.find_by_purls(gem_path, gem_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", gem_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Ruby gems: {e}"); - } - } - } - } - } - - // golang - #[cfg(feature = "golang")] - if let Some(golang_purls) = partitioned.get(&Ecosystem::Golang) { - if !golang_purls.is_empty() { - let go_crawler = GoCrawler; - match go_crawler.get_module_cache_paths(options).await { - Ok(cache_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = cache_paths.first() { - println!("Using Go module cache at: {}", first.display()); - } - } - for cache_path in &cache_paths { - match go_crawler.find_by_purls(cache_path, golang_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", cache_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Go modules: {e}"); - } - } - } - } - } - - // maven — experimental, double-gated. See `maven_runtime_enabled`. - #[cfg(feature = "maven")] - if let Some(maven_purls) = partitioned.get(&Ecosystem::Maven) { - if !maven_purls.is_empty() && !maven_runtime_enabled() { - if !silent { - warn_maven_disabled(maven_purls.len()); - } - } else if !maven_purls.is_empty() { - let maven_crawler = MavenCrawler; - match maven_crawler.get_maven_repo_paths(options).await { - Ok(repo_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = repo_paths.first() { - println!("Using Maven repository at: {}", first.display()); - } - } - for repo_path in &repo_paths { - match maven_crawler.find_by_purls(repo_path, maven_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", repo_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find Maven packages: {e}"); - } - } - } - } - } - - // composer - #[cfg(feature = "composer")] - if let Some(composer_purls) = partitioned.get(&Ecosystem::Composer) { - if !composer_purls.is_empty() { - let composer_crawler = ComposerCrawler; - match composer_crawler.get_vendor_paths(options).await { - Ok(vendor_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = vendor_paths.first() { - println!("Using PHP vendor packages at: {}", first.display()); - } - } - for vendor_path in &vendor_paths { - match composer_crawler.find_by_purls(vendor_path, composer_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", vendor_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find PHP packages: {e}"); - } - } - } - } - } - - // nuget — experimental, double-gated. See `nuget_runtime_enabled`. - #[cfg(feature = "nuget")] - if let Some(nuget_purls) = partitioned.get(&Ecosystem::Nuget) { - if !nuget_purls.is_empty() && !nuget_runtime_enabled() { - if !silent { - warn_nuget_disabled(nuget_purls.len()); - } - } else if !nuget_purls.is_empty() { - let nuget_crawler = NuGetCrawler; - match nuget_crawler.get_nuget_package_paths(options).await { - Ok(pkg_paths) => { - if (options.global || options.global_prefix.is_some()) && !silent { - if let Some(first) = pkg_paths.first() { - println!("Using NuGet packages at: {}", first.display()); - } - } - for pkg_path in &pkg_paths { - match nuget_crawler.find_by_purls(pkg_path, nuget_purls).await { - Ok(packages) => { - for (purl, pkg) in packages { - all_packages.entry(purl).or_insert(pkg.path); - } - } - Err(e) => { - if !silent { - eprintln!("Warning: Failed to scan {}: {}", pkg_path.display(), e); - } - } - } - } - } - Err(e) => { - if !silent { - eprintln!("Failed to find NuGet packages: {e}"); - } - } - } - } - } - - all_packages -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/socket-patch-core/src/api/blob_fetcher.rs b/crates/socket-patch-core/src/api/blob_fetcher.rs index fca94451..cf96a1d5 100644 --- a/crates/socket-patch-core/src/api/blob_fetcher.rs +++ b/crates/socket-patch-core/src/api/blob_fetcher.rs @@ -54,7 +54,7 @@ pub struct BlobFetchResult { } /// Aggregate result of a blob-fetch operation. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub struct FetchMissingBlobsResult { pub total: usize, pub downloaded: usize, @@ -114,40 +114,44 @@ pub async fn fetch_missing_blobs( let missing = get_missing_blobs(manifest, blobs_path).await; if missing.is_empty() { - return FetchMissingBlobsResult { - total: 0, - downloaded: 0, - failed: 0, - skipped: 0, - results: Vec::new(), - }; + return FetchMissingBlobsResult::default(); } // Ensure blobs directory exists if let Err(e) = tokio::fs::create_dir_all(blobs_path).await { - // If we cannot create the directory, every blob will fail. - let results: Vec = missing - .iter() - .map(|h| BlobFetchResult { - hash: h.clone(), - success: false, - error: Some(format!("Cannot create blobs directory: {}", e)), - }) - .collect(); - let failed = results.len(); - return FetchMissingBlobsResult { - total: failed, - downloaded: 0, - failed, - skipped: 0, - results, - }; + return all_failed_result(missing.iter(), |h| { + (h.clone(), format!("Cannot create blobs directory: {}", e)) + }); } let hashes: Vec = missing.into_iter().collect(); download_hashes(&hashes, blobs_path, client, on_progress).await } +/// Build a [`FetchMissingBlobsResult`] whose entries are all failures +/// for the same reason. Used by the early-return branches that hit a +/// blocker (e.g. cannot create blobs dir) before any download attempt. +fn all_failed_result<'a, I, F>(items: I, mut into_pair: F) -> FetchMissingBlobsResult +where + I: IntoIterator, + F: FnMut(&'a String) -> (String, String), +{ + let results: Vec = items + .into_iter() + .map(|item| { + let (hash, error) = into_pair(item); + BlobFetchResult { hash, success: false, error: Some(error) } + }) + .collect(); + let failed = results.len(); + FetchMissingBlobsResult { + total: failed, + failed, + results, + ..FetchMissingBlobsResult::default() + } +} + /// Download specific blobs identified by their hashes. /// /// Useful for fetching `beforeHash` blobs during rollback, where only a @@ -161,33 +165,14 @@ pub async fn fetch_blobs_by_hash( on_progress: Option<&OnProgress>, ) -> FetchMissingBlobsResult { if hashes.is_empty() { - return FetchMissingBlobsResult { - total: 0, - downloaded: 0, - failed: 0, - skipped: 0, - results: Vec::new(), - }; + return FetchMissingBlobsResult::default(); } // Ensure blobs directory exists if let Err(e) = tokio::fs::create_dir_all(blobs_path).await { - let results: Vec = hashes - .iter() - .map(|h| BlobFetchResult { - hash: h.clone(), - success: false, - error: Some(format!("Cannot create blobs directory: {}", e)), - }) - .collect(); - let failed = results.len(); - return FetchMissingBlobsResult { - total: failed, - downloaded: 0, - failed, - skipped: 0, - results, - }; + return all_failed_result(hashes.iter(), |h| { + (h.clone(), format!("Cannot create blobs directory: {}", e)) + }); } // Filter out hashes that already exist on disk @@ -281,7 +266,7 @@ pub async fn fetch_missing_sources( fetch_missing_archives_inner(manifest, dir, ArchiveKind::Diff, client, on_progress) .await } - None => empty_result(), + None => FetchMissingBlobsResult::default(), }, DownloadMode::Package => match sources.packages_path { Some(dir) => fetch_missing_archives_inner( @@ -292,7 +277,7 @@ pub async fn fetch_missing_sources( on_progress, ) .await, - None => empty_result(), + None => FetchMissingBlobsResult::default(), }, } } @@ -303,16 +288,6 @@ enum ArchiveKind { Package, } -fn empty_result() -> FetchMissingBlobsResult { - FetchMissingBlobsResult { - total: 0, - downloaded: 0, - failed: 0, - skipped: 0, - results: Vec::new(), - } -} - async fn fetch_missing_archives_inner( manifest: &PatchManifest, archives_dir: &Path, @@ -322,26 +297,13 @@ async fn fetch_missing_archives_inner( ) -> FetchMissingBlobsResult { let missing = get_missing_archives(manifest, archives_dir).await; if missing.is_empty() { - return empty_result(); + return FetchMissingBlobsResult::default(); } if let Err(e) = tokio::fs::create_dir_all(archives_dir).await { - let results: Vec = missing - .iter() - .map(|u| BlobFetchResult { - hash: u.clone(), - success: false, - error: Some(format!("Cannot create archives directory: {}", e)), - }) - .collect(); - let failed = results.len(); - return FetchMissingBlobsResult { - total: failed, - downloaded: 0, - failed, - skipped: 0, - results, - }; + return all_failed_result(missing.iter(), |u| { + (u.clone(), format!("Cannot create archives directory: {}", e)) + }); } let uuids: Vec = missing.into_iter().collect(); diff --git a/crates/socket-patch-core/src/api/client.rs b/crates/socket-patch-core/src/api/client.rs index a9975104..a1dfe50f 100644 --- a/crates/socket-patch-core/src/api/client.rs +++ b/crates/socket-patch-core/src/api/client.rs @@ -230,20 +230,23 @@ impl ApiClient { self.get_json(&path).await } - /// Search patches by CVE ID. - pub async fn search_patches_by_cve( + /// Shared implementation for `search_patches_by_{cve,ghsa,package}`. + /// `route` is the `by-` URL segment — the rest of the path layout + /// is identical across the three endpoints. + async fn search_patches_by_route( &self, org_slug: Option<&str>, - cve_id: &str, + route: &str, + identifier: &str, ) -> Result { - let encoded = urlencoding_encode(cve_id); + let encoded = urlencoding_encode(identifier); let path = if self.use_public_proxy { - format!("/patch/by-cve/{}", encoded) + format!("/patch/{route}/{encoded}") } else { let slug = org_slug .or(self.org_slug.as_deref()) .unwrap_or("default"); - format!("/v0/orgs/{}/patches/by-cve/{}", slug, encoded) + format!("/v0/orgs/{slug}/patches/{route}/{encoded}") }; let result = self.get_json::(&path).await?; Ok(result.unwrap_or_else(|| SearchResponse { @@ -252,26 +255,22 @@ impl ApiClient { })) } + /// Search patches by CVE ID. + pub async fn search_patches_by_cve( + &self, + org_slug: Option<&str>, + cve_id: &str, + ) -> Result { + self.search_patches_by_route(org_slug, "by-cve", cve_id).await + } + /// Search patches by GHSA ID. pub async fn search_patches_by_ghsa( &self, org_slug: Option<&str>, ghsa_id: &str, ) -> Result { - let encoded = urlencoding_encode(ghsa_id); - let path = if self.use_public_proxy { - format!("/patch/by-ghsa/{}", encoded) - } else { - let slug = org_slug - .or(self.org_slug.as_deref()) - .unwrap_or("default"); - format!("/v0/orgs/{}/patches/by-ghsa/{}", slug, encoded) - }; - let result = self.get_json::(&path).await?; - Ok(result.unwrap_or_else(|| SearchResponse { - patches: Vec::new(), - can_access_paid_patches: false, - })) + self.search_patches_by_route(org_slug, "by-ghsa", ghsa_id).await } /// Search patches by package PURL. @@ -283,20 +282,7 @@ impl ApiClient { org_slug: Option<&str>, purl: &str, ) -> Result { - let encoded = urlencoding_encode(purl); - let path = if self.use_public_proxy { - format!("/patch/by-package/{}", encoded) - } else { - let slug = org_slug - .or(self.org_slug.as_deref()) - .unwrap_or("default"); - format!("/v0/orgs/{}/patches/by-package/{}", slug, encoded) - }; - let result = self.get_json::(&path).await?; - Ok(result.unwrap_or_else(|| SearchResponse { - patches: Vec::new(), - can_access_paid_patches: false, - })) + self.search_patches_by_route(org_slug, "by-package", purl).await } /// Search patches for multiple packages (batch). diff --git a/crates/socket-patch-core/src/utils/cleanup_blobs.rs b/crates/socket-patch-core/src/utils/cleanup_blobs.rs index 118f55bb..0122aec9 100644 --- a/crates/socket-patch-core/src/utils/cleanup_blobs.rs +++ b/crates/socket-patch-core/src/utils/cleanup_blobs.rs @@ -13,81 +13,73 @@ pub struct CleanupResult { pub removed_blobs: Vec, } -/// Cleans up unused blob files from the blobs directory. -/// -/// Analyzes the manifest to determine which afterHash blobs are needed for applying patches, -/// then removes any blob files that are not needed. +/// Shared core for `cleanup_unused_blobs` / `cleanup_unused_archives`. /// -/// Note: beforeHash blobs are considered "unused" because they are downloaded on-demand -/// during rollback operations. This saves disk space since beforeHash blobs are only -/// needed for rollback, not for applying patches. -pub async fn cleanup_unused_blobs( - manifest: &PatchManifest, - blobs_dir: &Path, +/// Walks `dir`, treats it as authoritative socket-patch state (so any +/// regular non-hidden file is considered for removal), and asks +/// `is_used(filename) -> bool` whether each file should be kept. +async fn cleanup_dir bool>( + dir: &Path, dry_run: bool, + is_used: F, ) -> Result { - // Only keep afterHash blobs - beforeHash blobs are downloaded on-demand during rollback - let used_blobs = get_after_hash_blobs(manifest); - - // Check if blobs directory exists - if tokio::fs::metadata(blobs_dir).await.is_err() { - // Blobs directory doesn't exist, nothing to clean up - return Ok(CleanupResult { - blobs_checked: 0, - blobs_removed: 0, - bytes_freed: 0, - removed_blobs: vec![], - }); + if tokio::fs::metadata(dir).await.is_err() { + return Ok(CleanupResult::default()); } - // Read all files in the blobs directory - let mut read_dir = tokio::fs::read_dir(blobs_dir).await?; - let mut blob_entries = Vec::new(); - + let mut read_dir = tokio::fs::read_dir(dir).await?; + let mut entries = Vec::new(); while let Some(entry) = read_dir.next_entry().await? { - blob_entries.push(entry); + entries.push(entry); } let mut result = CleanupResult { - blobs_checked: blob_entries.len(), - blobs_removed: 0, - bytes_freed: 0, - removed_blobs: vec![], + blobs_checked: entries.len(), + ..CleanupResult::default() }; - // Check each blob file - for entry in &blob_entries { - let file_name = entry.file_name(); - let file_name_str = file_name.to_string_lossy().to_string(); - - // Skip hidden files and directories + for entry in &entries { + let file_name_str = entry.file_name().to_string_lossy().to_string(); if file_name_str.starts_with('.') { continue; } - - let blob_path = blobs_dir.join(&file_name_str); - - // Check if it's a file (not a directory) - let metadata = tokio::fs::metadata(&blob_path).await?; + let path = dir.join(&file_name_str); + let metadata = tokio::fs::metadata(&path).await?; if !metadata.is_file() { continue; } - - // If this blob is not in use, remove it - if !used_blobs.contains(&file_name_str) { - result.blobs_removed += 1; - result.bytes_freed += metadata.len(); - result.removed_blobs.push(file_name_str); - - if !dry_run { - tokio::fs::remove_file(&blob_path).await?; - } + if is_used(&file_name_str) { + continue; + } + result.blobs_removed += 1; + result.bytes_freed += metadata.len(); + result.removed_blobs.push(file_name_str); + if !dry_run { + tokio::fs::remove_file(&path).await?; } } Ok(result) } +/// Cleans up unused blob files from the blobs directory. +/// +/// Analyzes the manifest to determine which afterHash blobs are needed for applying patches, +/// then removes any blob files that are not needed. +/// +/// Note: beforeHash blobs are considered "unused" because they are downloaded on-demand +/// during rollback operations. This saves disk space since beforeHash blobs are only +/// needed for rollback, not for applying patches. +pub async fn cleanup_unused_blobs( + manifest: &PatchManifest, + blobs_dir: &Path, + dry_run: bool, +) -> Result { + // Only keep afterHash blobs - beforeHash blobs are downloaded on-demand during rollback + let used_blobs = get_after_hash_blobs(manifest); + cleanup_dir(blobs_dir, dry_run, |name| used_blobs.contains(name)).await +} + /// Cleans up unused per-patch archive files from `archives_dir`. /// /// Archives are named `.tar.gz`. Any file matching that @@ -102,62 +94,15 @@ pub async fn cleanup_unused_archives( archives_dir: &Path, dry_run: bool, ) -> Result { - let used_uuids: HashSet = manifest - .patches - .values() - .map(|r| r.uuid.clone()) - .collect(); - - if tokio::fs::metadata(archives_dir).await.is_err() { - return Ok(CleanupResult { - blobs_checked: 0, - blobs_removed: 0, - bytes_freed: 0, - removed_blobs: vec![], - }); - } - - let mut read_dir = tokio::fs::read_dir(archives_dir).await?; - let mut entries = Vec::new(); - while let Some(entry) = read_dir.next_entry().await? { - entries.push(entry); - } - - let mut result = CleanupResult { - blobs_checked: entries.len(), - blobs_removed: 0, - bytes_freed: 0, - removed_blobs: vec![], - }; - - for entry in &entries { - let file_name = entry.file_name(); - let file_name_str = file_name.to_string_lossy().to_string(); - if file_name_str.starts_with('.') { - continue; - } - let archive_path = archives_dir.join(&file_name_str); - let metadata = tokio::fs::metadata(&archive_path).await?; - if !metadata.is_file() { - continue; - } - // Strip the .tar.gz suffix to recover the UUID; if it doesn't end - // in .tar.gz, treat the entry as orphaned and remove it. - let uuid_part = file_name_str - .strip_suffix(".tar.gz") - .unwrap_or(&file_name_str); - if used_uuids.contains(uuid_part) { - continue; - } - result.blobs_removed += 1; - result.bytes_freed += metadata.len(); - result.removed_blobs.push(file_name_str); - if !dry_run { - tokio::fs::remove_file(&archive_path).await?; - } - } - - Ok(result) + let used_uuids: HashSet = + manifest.patches.values().map(|r| r.uuid.clone()).collect(); + cleanup_dir(archives_dir, dry_run, |name| { + // Strip the .tar.gz suffix to recover the UUID; if it doesn't + // end in .tar.gz, treat the entry as orphaned (not "used"). + let uuid_part = name.strip_suffix(".tar.gz").unwrap_or(name); + used_uuids.contains(uuid_part) + }) + .await } /// Formats the cleanup result for human-readable output. diff --git a/crates/socket-patch-core/src/utils/telemetry.rs b/crates/socket-patch-core/src/utils/telemetry.rs index d67d2510..207b1a48 100644 --- a/crates/socket-patch-core/src/utils/telemetry.rs +++ b/crates/socket-patch-core/src/utils/telemetry.rs @@ -359,6 +359,44 @@ pub async fn track_patch_event(options: TrackPatchEventOptions) { // convenient (callers typically have `Option` and call `.as_deref()`). // --------------------------------------------------------------------------- +/// Convert a `serde_json::json!({...})` object into the `HashMap` that +/// [`TrackPatchEventOptions::metadata`] expects, swallowing the conversion +/// to avoid `.unwrap()` noise at every call site. +fn metadata_from_json(value: serde_json::Value) -> Option> { + match value { + serde_json::Value::Object(map) => { + if map.is_empty() { + None + } else { + Some(map.into_iter().collect()) + } + } + _ => None, + } +} + +/// Shared fire-and-forget helper for the per-event tracker wrappers below. +/// Centralizes the `String::from` plumbing for the four optional fields +/// that every tracker shares. +async fn fire( + event_type: PatchTelemetryEventType, + command: &'static str, + metadata: serde_json::Value, + error: Option, + api_token: Option<&str>, + org_slug: Option<&str>, +) { + track_patch_event(TrackPatchEventOptions { + event_type, + command: command.to_string(), + metadata: metadata_from_json(metadata), + error: error.map(|e| ("Error".to_string(), e.to_string())), + api_token: api_token.map(String::from), + org_slug: org_slug.map(String::from), + }) + .await; +} + /// Track a successful patch application. pub async fn track_patch_applied( patches_count: usize, @@ -366,21 +404,14 @@ pub async fn track_patch_applied( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "patches_count".to_string(), - serde_json::Value::Number(serde_json::Number::from(patches_count)), - ); - metadata.insert("dry_run".to_string(), serde_json::Value::Bool(dry_run)); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchApplied, - command: "apply".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchApplied, + "apply", + serde_json::json!({ "patches_count": patches_count, "dry_run": dry_run }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -394,17 +425,14 @@ pub async fn track_patch_apply_failed( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert("dry_run".to_string(), serde_json::Value::Bool(dry_run)); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchApplyFailed, - command: "apply".to_string(), - metadata: Some(metadata), - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchApplyFailed, + "apply", + serde_json::json!({ "dry_run": dry_run }), + Some(error), + api_token, + org_slug, + ) .await; } @@ -414,39 +442,31 @@ pub async fn track_patch_removed( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "removed_count".to_string(), - serde_json::Value::Number(serde_json::Number::from(removed_count)), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchRemoved, - command: "remove".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchRemoved, + "remove", + serde_json::json!({ "removed_count": removed_count }), + None::<&str>, + api_token, + org_slug, + ) .await; } -/// Track a failed patch removal. -/// -/// Accepts any `Display` type for the error. +/// Track a failed patch removal. Accepts any `Display` type for the error. pub async fn track_patch_remove_failed( error: impl std::fmt::Display, api_token: Option<&str>, org_slug: Option<&str>, ) { - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchRemoveFailed, - command: "remove".to_string(), - metadata: None, - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchRemoveFailed, + "remove", + serde_json::Value::Null, + Some(error), + api_token, + org_slug, + ) .await; } @@ -456,39 +476,31 @@ pub async fn track_patch_rolled_back( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "rolled_back_count".to_string(), - serde_json::Value::Number(serde_json::Number::from(rolled_back_count)), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchRolledBack, - command: "rollback".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchRolledBack, + "rollback", + serde_json::json!({ "rolled_back_count": rolled_back_count }), + None::<&str>, + api_token, + org_slug, + ) .await; } -/// Track a failed patch rollback. -/// -/// Accepts any `Display` type for the error. +/// Track a failed patch rollback. Accepts any `Display` type for the error. pub async fn track_patch_rollback_failed( error: impl std::fmt::Display, api_token: Option<&str>, org_slug: Option<&str>, ) { - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchRollbackFailed, - command: "rollback".to_string(), - metadata: None, - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchRollbackFailed, + "rollback", + serde_json::Value::Null, + Some(error), + api_token, + org_slug, + ) .await; } @@ -516,45 +528,21 @@ pub async fn track_patch_scanned( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "packages_scanned".to_string(), - serde_json::Value::Number(serde_json::Number::from(packages_scanned)), - ); - metadata.insert( - "free_patches".to_string(), - serde_json::Value::Number(serde_json::Number::from(free_patches)), - ); - metadata.insert( - "paid_patches".to_string(), - serde_json::Value::Number(serde_json::Number::from(paid_patches)), - ); - metadata.insert( - "can_access_paid".to_string(), - serde_json::Value::Bool(can_access_paid), - ); - metadata.insert( - "ecosystems".to_string(), - serde_json::Value::Array( - ecosystems - .iter() - .map(|e| serde_json::Value::String(e.clone())) - .collect(), - ), - ); - metadata.insert( - "fallback_to_proxy".to_string(), - serde_json::Value::Bool(fallback_to_proxy), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchScanned, - command: "scan".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchScanned, + "scan", + serde_json::json!({ + "packages_scanned": packages_scanned, + "free_patches": free_patches, + "paid_patches": paid_patches, + "can_access_paid": can_access_paid, + "ecosystems": ecosystems, + "fallback_to_proxy": fallback_to_proxy, + }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -565,20 +553,14 @@ pub async fn track_patch_scan_failed( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "fallback_to_proxy".to_string(), - serde_json::Value::Bool(fallback_to_proxy), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchScanFailed, - command: "scan".to_string(), - metadata: Some(metadata), - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchScanFailed, + "scan", + serde_json::json!({ "fallback_to_proxy": fallback_to_proxy }), + Some(error), + api_token, + org_slug, + ) .await; } @@ -594,36 +576,20 @@ pub async fn track_patch_fetched( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "uuid".to_string(), - serde_json::Value::String(uuid.to_string()), - ); - metadata.insert( - "tier".to_string(), - serde_json::Value::String(tier.to_string()), - ); - metadata.insert( - "ecosystem".to_string(), - serde_json::Value::String(ecosystem.to_string()), - ); - metadata.insert( - "download_mode".to_string(), - serde_json::Value::String(download_mode.to_string()), - ); - metadata.insert( - "fallback_to_proxy".to_string(), - serde_json::Value::Bool(fallback_to_proxy), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchFetched, - command: "get".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchFetched, + "get", + serde_json::json!({ + "uuid": uuid, + "tier": tier, + "ecosystem": ecosystem, + "download_mode": download_mode, + "fallback_to_proxy": fallback_to_proxy, + }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -636,24 +602,14 @@ pub async fn track_patch_fetch_failed( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "uuid".to_string(), - serde_json::Value::String(uuid.to_string()), - ); - metadata.insert( - "fallback_to_proxy".to_string(), - serde_json::Value::Bool(fallback_to_proxy), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchFetchFailed, - command: "get".to_string(), - metadata: Some(metadata), - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchFetchFailed, + "get", + serde_json::json!({ "uuid": uuid, "fallback_to_proxy": fallback_to_proxy }), + Some(error), + api_token, + org_slug, + ) .await; } @@ -667,20 +623,14 @@ pub async fn track_patch_listed( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "patches_count".to_string(), - serde_json::Value::Number(serde_json::Number::from(patches_count)), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchListed, - command: "list".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchListed, + "list", + serde_json::json!({ "patches_count": patches_count }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -692,28 +642,18 @@ pub async fn track_patch_repaired( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "blobs_added".to_string(), - serde_json::Value::Number(serde_json::Number::from(blobs_added)), - ); - metadata.insert( - "blobs_removed".to_string(), - serde_json::Value::Number(serde_json::Number::from(blobs_removed)), - ); - metadata.insert( - "bytes_freed".to_string(), - serde_json::Value::Number(serde_json::Number::from(bytes_freed)), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchRepaired, - command: "repair".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchRepaired, + "repair", + serde_json::json!({ + "blobs_added": blobs_added, + "blobs_removed": blobs_removed, + "bytes_freed": bytes_freed, + }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -723,14 +663,14 @@ pub async fn track_patch_repair_failed( api_token: Option<&str>, org_slug: Option<&str>, ) { - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchRepairFailed, - command: "repair".to_string(), - metadata: None, - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchRepairFailed, + "repair", + serde_json::Value::Null, + Some(error), + api_token, + org_slug, + ) .await; } @@ -741,20 +681,14 @@ pub async fn track_patch_setup( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "manager".to_string(), - serde_json::Value::String(manager.to_string()), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchSetup, - command: "setup".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchSetup, + "setup", + serde_json::json!({ "manager": manager }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -767,18 +701,14 @@ pub async fn track_patch_unlocked( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert("was_held".to_string(), serde_json::Value::Bool(was_held)); - metadata.insert("released".to_string(), serde_json::Value::Bool(released)); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchUnlocked, - command: "unlock".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchUnlocked, + "unlock", + serde_json::json!({ "was_held": was_held, "released": released }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -788,14 +718,14 @@ pub async fn track_patch_unlock_failed( api_token: Option<&str>, org_slug: Option<&str>, ) { - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::PatchUnlockFailed, - command: "unlock".to_string(), - metadata: None, - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::PatchUnlockFailed, + "unlock", + serde_json::Value::Null, + Some(error), + api_token, + org_slug, + ) .await; } @@ -812,28 +742,18 @@ pub async fn track_vex_generated( api_token: Option<&str>, org_slug: Option<&str>, ) { - let mut metadata = HashMap::new(); - metadata.insert( - "advisories_count".to_string(), - serde_json::Value::Number(serde_json::Number::from(advisories_count)), - ); - metadata.insert( - "format".to_string(), - serde_json::Value::String(format.to_string()), - ); - metadata.insert( - "output_kind".to_string(), - serde_json::Value::String(output_kind.to_string()), - ); - - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::VexGenerated, - command: "vex".to_string(), - metadata: Some(metadata), - error: None, - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::VexGenerated, + "vex", + serde_json::json!({ + "advisories_count": advisories_count, + "format": format, + "output_kind": output_kind, + }), + None::<&str>, + api_token, + org_slug, + ) .await; } @@ -843,14 +763,14 @@ pub async fn track_vex_failed( api_token: Option<&str>, org_slug: Option<&str>, ) { - track_patch_event(TrackPatchEventOptions { - event_type: PatchTelemetryEventType::VexFailed, - command: "vex".to_string(), - metadata: None, - error: Some(("Error".to_string(), error.to_string())), - api_token: api_token.map(|s| s.to_string()), - org_slug: org_slug.map(|s| s.to_string()), - }) + fire( + PatchTelemetryEventType::VexFailed, + "vex", + serde_json::Value::Null, + Some(error), + api_token, + org_slug, + ) .await; } From 596d5191c79814266f90ea95026fc7b479357251 Mon Sep 17 00:00:00 2001 From: Mikola Lysenko Date: Wed, 27 May 2026 15:51:33 -0400 Subject: [PATCH 2/2] refactor(ecosystem_dispatch): hoist MergeFn type alias for clippy `-D clippy::type_complexity` rejected the inline fn-pointer signature on `dispatch_find`'s `pypi_merge` argument. Lift it to a `MergeFn` type alias shared by `merge_first_wins` and `merge_pypi_qualified`. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/socket-patch-cli/src/ecosystem_dispatch.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/crates/socket-patch-cli/src/ecosystem_dispatch.rs b/crates/socket-patch-cli/src/ecosystem_dispatch.rs index 444b8692..02bc9562 100644 --- a/crates/socket-patch-cli/src/ecosystem_dispatch.rs +++ b/crates/socket-patch-cli/src/ecosystem_dispatch.rs @@ -155,6 +155,12 @@ macro_rules! scan_ecosystem { }}; } +/// Signature shared by `merge_first_wins` and `merge_pypi_qualified`. +/// `dispatch_find` swaps between them so the rollback path can fan one +/// crawler result back out to every caller-supplied qualified PURL. +type MergeFn = + fn(&mut HashMap, &[String], HashMap); + /// Default merge: insert the crawler-returned PURL → first wins. fn merge_first_wins( out: &mut HashMap, @@ -208,11 +214,7 @@ async fn dispatch_find( partitioned: &HashMap>, options: &CrawlerOptions, silent: bool, - pypi_merge: fn( - &mut HashMap, - &[String], - HashMap, - ), + pypi_merge: MergeFn, ) -> HashMap { let mut out: HashMap = HashMap::new();