diff --git a/Cargo.lock b/Cargo.lock index fd0ff85..fb30fe9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1783,7 +1783,7 @@ dependencies = [ ] [[package]] -name = "imojefs" +name = "imoje" version = "0.0.1" dependencies = [ "aes-gcm", @@ -1792,15 +1792,21 @@ dependencies = [ "aws-credential-types", "aws-sdk-s3", "base64 0.22.1", + "bytes", "clap", "dirs", "env_logger", + "hex", "lazy_static", "log", + "md5", + "mime-type", + "pluralizer", "rand 0.8.8", "rustsf", "serde", "serde_json", + "sha2 0.10.9", "signal-hook", "tokio", "toml", @@ -2040,6 +2046,12 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "md5" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ebb8d8732c6a6df3d8f032a82911cfc747e00efb95cc46e8d0acd5b5b88570c" + [[package]] name = "memchr" version = "2.8.3" @@ -2052,6 +2064,12 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "mime-type" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e3e3895477670a13f7d4bd8c675538c1a45cc222161809aef077ce2b2aa4174" + [[package]] name = "miniz_oxide" version = "0.9.1" @@ -2221,6 +2239,16 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" +[[package]] +name = "pluralizer" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35e4616e94b67b8b61846ea69d4bf041a62147d569d16f437689229e2677d38c" +dependencies = [ + "lazy_static", + "regex", +] + [[package]] name = "polyval" version = "0.6.2" @@ -2667,7 +2695,7 @@ dependencies = [ [[package]] name = "rustsf" -version = "0.0.6" +version = "0.1.0" dependencies = [ "anyhow", "base64 0.23.1", @@ -2687,8 +2715,6 @@ dependencies = [ [[package]] name = "rustsf_marcos" version = "0.0.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93d3322efe3680c19b17883c5bd67a2188a217568b449d9325919be16581013a" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 01bc29b..2dce85b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "imojefs" +name = "imoje" description = "Johnson & Johnson IMoje CLI" authors = ["Wim Velzeboer "] version = "0.0.1" @@ -12,15 +12,21 @@ aws-config = { version = "1.1.7", features = ["behavior-version-latest"] } aws-credential-types = "1.2.14" aws-sdk-s3 = "1.132.0" base64 = "0.22" +bytes = "1.12.1" clap = { version = "4.6.1", features = ["derive", "env"] } dirs = "6.0" env_logger = "0.11.10" +hex = "0.4" lazy_static = "1.5.0" log = "0.4" +md5 = "0.8.1" +mime-type = "0.2.0" +pluralizer = "0.4.0" rand = "0.8" rustsf = { path = "../rustsf" } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0.149" +sha2 = "0.10" signal-hook = "0.4.4" tokio = { version = "1", features = ["full"] } toml = "0.9" diff --git a/src/aws_s3.rs b/src/aws_s3.rs deleted file mode 100644 index 3446ced..0000000 --- a/src/aws_s3.rs +++ /dev/null @@ -1,34 +0,0 @@ -use aws_config::Region; - -use aws_credential_types::provider::SharedCredentialsProvider; - -pub(crate) async fn connect( - client_id: &str, - client_secret: &str, - custom_endpoint: Option<&str>, - region: &str) -> aws_sdk_s3::Client { - - let creds = SharedCredentialsProvider::new(aws_credential_types::Credentials::new( - client_id, client_secret, None, None, "Static", - )); - - let config = match custom_endpoint { - Some(endpoint) => { - aws_config::from_env() - .region(Region::new(region.to_string())) - .endpoint_url(endpoint) - .credentials_provider(creds) - .load() - .await - } - None => { - aws_config::from_env() - .region(Region::new(region.to_string())) - .credentials_provider(creds) - .load() - .await - } - }; - - aws_sdk_s3::Client::new(&config) -} \ No newline at end of file diff --git a/src/cli/mod.rs b/src/cli/mod.rs index 5cb40b5..acb3afc 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -8,9 +8,6 @@ mod set; #[derive(Parser, Debug)] #[command(version, about, long_about = None)] struct Cli { - /// Number of seconds to wait - #[arg(short = 'w', long, default_value_t = 20, env = "IMOJEFS_WAIT")] - pub wait: u64, /// The logging level #[arg(short = 'l', long, value_enum, default_value_t = LogLevel::Info, env = "IMOJEFS_LOG_LEVEL" @@ -27,7 +24,8 @@ struct Cli { #[derive(clap::Subcommand, Debug)] enum Commands { - /// Controls testing features + + /// Extracts binary contents from Salesforce, uploads it to a S3 bucket, and leaves a link in Salesforce FileSync, /// Lists the settings from the configuration file @@ -37,37 +35,41 @@ enum Commands { Set { /// A comma separated list of fallback email address, /// used when Salesforce cannot be reached to deliver log messages - #[arg(short = 'a', long, env = "IMOJEFS_ADMIN_EMAIL")] + #[arg(short = 'a', long, env = "IMOJE_ADMIN_EMAIL")] admin_email: Option, /// AWS S3 bucket name - #[arg(short = 'b', long = "s3-bucket", env = "IMOJEFS_S3_BUCKET")] + #[arg(short = 'b', long = "s3-bucket", env = "IMOJE_S3_BUCKET")] s3_bucket: Option, /// AWS Access Key Id - #[arg(short = 'k', long, env = "IMOJEFS_S3_CLIENT_ID")] + #[arg(short = 'k', long, env = "IMOJE_S3_CLIENT_ID")] s3_client_id: Option, /// AWS Secret Access Key - #[arg(short = 's', long, env = "IMOJEFS_S3_CLIENT_SECRET", hide_env_values = true)] + #[arg(short = 's', long, env = "IMOJE_S3_CLIENT_SECRET", hide_env_values = true)] s3_client_secret: Option, /// AWS Endpoint URL, [default: AWS url] - #[arg(short = 'e', long, env = "IMOJEFS_S3_ENDPOINT")] + #[arg(short = 'e', long, env = "IMOJE_S3_ENDPOINT")] s3_endpoint: Option, /// AWS Region - #[arg(short = 'r', long, default_value = "us-east-1", env = "IMOJEFS_S3_REGION")] + #[arg(short = 'r', long, default_value = "us-east-1", env = "IMOJE_S3_REGION")] s3_region: Option, /// Salesforce Authentication Url #[arg( short = 'x', long = "sfdx-auth-url", - env = "IMOJEFS_SFDX_AUTH_URL", + env = "IMOJE_SFDX_AUTH_URL", hide_env_values = true )] sfdx_auth_url: Option, + + /// Number of seconds to wait between each poll + #[arg(short = 'w', long, default_value_t = 20, env = "IMOJE_WAIT")] + wait: u64, }, } @@ -103,6 +105,7 @@ pub(crate) fn run() { s3_endpoint, s3_region, sfdx_auth_url, + wait, } => run_set_command( &cli.config, admin_email, @@ -112,9 +115,14 @@ pub(crate) fn run() { s3_endpoint, s3_region, sfdx_auth_url, + wait, ), Commands::List => run_list_command(&cli), - Commands::FileSync { .. } => run_file_sync(&cli), + Commands::FileSync => run_file_sync(&cli) + .unwrap_or_else(|e| { + error!("Failed to run file sync: {:#}", e); + std::process::exit(1); + }) }, None => { println!("Expected subcommand, none given. Use --help for more information"); @@ -131,18 +139,24 @@ fn load_configuration_file(config_file: &str) -> Result<()> { Ok(()) } -fn run_file_sync(cli: &Cli) { +fn run_file_sync(cli: &Cli) -> Result<()> { logger::init(&cli.log_level); + load_configuration_file(&cli.config)?; - match load_configuration_file(&cli.config) { - Ok(()) => {} - Err(e) => { - error!("Error loading configuration file: {:#}", e); - std::process::exit(1); - } - } + let app_config = + crate::config::load(&cli.config) + .context("Failed to load application configuration")? + .new_file_sync() + .context("File sync configuration fault")?; - crate::file_sync::run(); + let rt = tokio::runtime::Runtime::new() + .context("Failed to create async runtime")?; + + rt.block_on(async { + crate::file_sync::run(app_config).await + })?; + + Ok(()) } fn run_list_command(cli: &Cli) { @@ -150,7 +164,7 @@ fn run_list_command(cli: &Cli) { .context("Failed to load application configuration") { Ok(()) => {} Err(e) => { - error!("Error listing configuration settings: {:#}", e); + println!("Error listing configuration settings: {:#}", e); std::process::exit(1); } } @@ -165,6 +179,7 @@ fn run_set_command( s3_endpoint: &Option, s3_region: &Option, sfdx_auth_url: &Option, + wait: &u64, ) { match set::run( config_file, @@ -175,6 +190,7 @@ fn run_set_command( s3_endpoint, s3_region, sfdx_auth_url, + wait ) { Ok(()) => { println!("Configuration settings updated successfully"); diff --git a/src/cli/set/mod.rs b/src/cli/set/mod.rs index 3338dcc..6228f7a 100644 --- a/src/cli/set/mod.rs +++ b/src/cli/set/mod.rs @@ -7,7 +7,9 @@ pub(crate) fn run(config_file: &str, s3_client_secret: &Option, s3_endpoint: &Option, s3_region: &Option, - sfdx_auth_url: &Option,) -> Result<()> { + sfdx_auth_url: &Option, + wait: &u64, +) -> Result<()> { let mut updated = false; @@ -50,6 +52,8 @@ pub(crate) fn run(config_file: &str, updated = true; } + app_config.wait = *wait; + crate::config::save(config_file, &app_config) .context("Failed to save configuration")?; diff --git a/src/config.rs b/src/config.rs index b1f689a..1fdd210 100644 --- a/src/config.rs +++ b/src/config.rs @@ -3,14 +3,14 @@ use std::path::PathBuf; use aes_gcm::aead::{Aead, KeyInit, OsRng}; use aes_gcm::{Aes256Gcm, Nonce}; -use anyhow::{anyhow, Context, Result}; -use base64::engine::general_purpose::STANDARD as BASE64; +use anyhow::{Context, Result, anyhow}; use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; use rand::RngCore; use serde::{Deserialize, Serialize}; const CONFIG_KEY_ENV: &str = "IMOJE_CONFIG_KEY"; -const CONFIG_DIR: &str = "config/imoje"; +const CONFIG_DIR: &str = ".config/imoje"; const NONCE_LEN: usize = 12; const NOT_SET: &'static str = ""; @@ -25,6 +25,7 @@ pub struct AppConfig { pub s3_endpoint: Option, pub s3_region: Option, pub sfdx_auth_url: Option, + pub wait: u64, } impl Default for AppConfig { @@ -37,23 +38,34 @@ impl Default for AppConfig { s3_endpoint: None, s3_region: None, sfdx_auth_url: None, + wait: 20, } } } +impl AppConfig { + pub(crate) fn new_file_sync(&self) -> Result { + let config = crate::file_sync::FileSyncConfig { + admin_email: self.admin_email.clone(), + s3_bucket: self.s3_bucket.clone().with_context(|| "S3 Bucket is required")?, + s3_client_id: self.s3_client_id.clone().with_context(|| "S3 Client ID is required")?, + s3_client_secret: self.s3_client_secret.clone().with_context(|| "S3 Client Secret is required")?, + s3_endpoint: self.s3_endpoint.clone(), + s3_region: self.s3_region.clone(), + sfdx_auth_url: self.sfdx_auth_url.clone().with_context(|| "SFDX Auth URL is required")?, + wait: self.wait, + }; + Ok(config) + } +} + pub fn list(config_file: &str) -> Result<()> { let app_config = load(config_file).context("Failed to load application configuration")?; println!("Configuration: {}", config_file); println!("----------------------------------------------------"); - println!( - "admin_email = {}", - app_config.admin_email.as_deref().unwrap_or(NOT_SET) - ); - println!( - "s3_bucket = {}", - app_config.s3_bucket.as_deref().unwrap_or(NOT_SET) - ); + println!("admin_email = {}", app_config.admin_email.as_deref().unwrap_or(NOT_SET)); + println!("s3_bucket = {}", app_config.s3_bucket.as_deref().unwrap_or(NOT_SET)); println!( "s3_client_id = {}", app_config.s3_client_id.as_deref().unwrap_or(NOT_SET) @@ -65,15 +77,9 @@ pub fn list(config_file: &str) -> Result<()> { None => NOT_SET, } ); - println!( - "s3_endpoint = {}", - app_config.s3_endpoint.as_deref().unwrap_or(NOT_SET) - ); + println!("s3_endpoint = {}", app_config.s3_endpoint.as_deref().unwrap_or(NOT_SET)); - println!( - "s3_region = {}", - app_config.s3_region.as_deref().unwrap_or(NOT_SET) - ); + println!("s3_region = {}", app_config.s3_region.as_deref().unwrap_or(NOT_SET)); println!( "sfdx_auth_url = {}", match app_config.sfdx_auth_url { @@ -81,9 +87,9 @@ pub fn list(config_file: &str) -> Result<()> { None => NOT_SET, } ); + println!("wait = {}", app_config.wait); Ok(()) - } pub fn load(config_file: &str) -> Result { @@ -122,11 +128,9 @@ fn save_encrypted(config_path: &PathBuf, config: &AppConfig) -> Result<()> { .with_context(|| format!("Unable to create configuration directory {}", parent.display()))?; } - let serialized_config = toml::to_string_pretty(config) - .context("Unable to serialize default configuration")?; + let serialized_config = toml::to_string_pretty(config).context("Unable to serialize default configuration")?; - let encrypted_config = encrypt_config(&serialized_config) - .context("Unable to encrypt default configuration")?; + let encrypted_config = encrypt_config(&serialized_config).context("Unable to encrypt default configuration")?; fs::write(config_path, encrypted_config) .with_context(|| format!("Unable to write configuration file {}", config_path.display()))?; @@ -136,8 +140,7 @@ fn save_encrypted(config_path: &PathBuf, config: &AppConfig) -> Result<()> { fn encrypt_config(config: &str) -> Result { let key = config_key()?; - let cipher = Aes256Gcm::new_from_slice(&key) - .map_err(|_| anyhow!("Invalid encryption key length"))?; + let cipher = Aes256Gcm::new_from_slice(&key).map_err(|_| anyhow!("Invalid encryption key length"))?; let mut nonce_bytes = [0_u8; NONCE_LEN]; OsRng.fill_bytes(&mut nonce_bytes); @@ -163,8 +166,7 @@ fn decrypt_config(encrypted_config: &str) -> Result { } let key = config_key()?; - let cipher = Aes256Gcm::new_from_slice(&key) - .map_err(|_| anyhow!("Invalid encryption key length"))?; + let cipher = Aes256Gcm::new_from_slice(&key).map_err(|_| anyhow!("Invalid encryption key length"))?; let nonce = Nonce::from_slice(&encrypted_payload[..NONCE_LEN]); let ciphertext = &encrypted_payload[NONCE_LEN..]; @@ -173,15 +175,14 @@ fn decrypt_config(encrypted_config: &str) -> Result { .decrypt(nonce, ciphertext) .map_err(|error| anyhow!("Unable to decrypt configuration: {error}"))?; - let plaintext = String::from_utf8(plaintext) - .context("Configuration file does not contain valid UTF-8")?; + let plaintext = String::from_utf8(plaintext).context("Configuration file does not contain valid UTF-8")?; toml::from_str(&plaintext).context("Unable to parse decrypted configuration") } fn config_key() -> Result<[u8; 32]> { - let encoded_key = std::env::var(CONFIG_KEY_ENV) - .unwrap_or("ihDlaCqJqUGJ1am4dT9k/BHRG2UT+iK9IfYcPe0Db/I=".to_string()); + let encoded_key = + std::env::var(CONFIG_KEY_ENV).unwrap_or("ihDlaCqJqUGJ1am4dT9k/BHRG2UT+iK9IfYcPe0Db/I=".to_string()); // .with_context(|| format!("Missing required environment variable {CONFIG_KEY_ENV}"))?; let key = BASE64 @@ -290,8 +291,7 @@ mod tests { let _lock = env_lock(); set_test_key(); - let encrypted = encrypt_config("sfdx_auth_url = 'auth-url'") - .expect("configuration should encrypt"); + let encrypted = encrypt_config("sfdx_auth_url = 'auth-url'").expect("configuration should encrypt"); let encrypted_payload = BASE64 .decode(encrypted) .expect("encrypted configuration should be base64"); @@ -339,6 +339,52 @@ s3_bucket = "example-bucket" assert!(error.to_string().contains("Configuration file is too short")); } + #[test] + fn new_file_sync_builds_file_sync_config_when_required_values_are_present() { + let config = AppConfig { + admin_email: Some("admin@example.com".to_string()), + s3_bucket: Some("example-bucket".to_string()), + s3_client_id: Some("client-id".to_string()), + s3_client_secret: Some("client-secret".to_string()), + s3_endpoint: Some("https://s3.example.com".to_string()), + s3_region: Some("us-east-1".to_string()), + sfdx_auth_url: Some("force://example".to_string()), + wait: 0, + }; + + let file_sync_config = config + .new_file_sync() + .expect("complete configuration should create file sync config"); + + assert_eq!(file_sync_config.admin_email.as_deref(), Some("admin@example.com")); + assert_eq!(file_sync_config.s3_bucket, "example-bucket"); + assert_eq!(file_sync_config.s3_client_id, "client-id"); + assert_eq!(file_sync_config.s3_client_secret, "client-secret"); + assert_eq!(file_sync_config.s3_endpoint.as_deref(), Some("https://s3.example.com")); + assert_eq!(file_sync_config.s3_region.as_deref(), Some("us-east-1")); + assert_eq!(file_sync_config.sfdx_auth_url, "force://example"); + } + + #[test] + fn new_file_sync_fails_when_required_value_is_missing() { + let config = AppConfig { + admin_email: Some("admin@example.com".to_string()), + s3_bucket: None, + s3_client_id: Some("client-id".to_string()), + s3_client_secret: Some("client-secret".to_string()), + s3_endpoint: None, + s3_region: Some("us-east-1".to_string()), + sfdx_auth_url: Some("force://example".to_string()), + wait: 0, + }; + + let error = config + .new_file_sync() + .expect_err("missing S3 bucket should fail"); + + assert!(error.to_string().contains("S3 Bucket is required")); + } + #[test] fn save_and_load_round_trip_configuration_file() { let _lock = env_lock(); @@ -354,6 +400,7 @@ s3_bucket = "example-bucket" s3_client_secret: None, s3_endpoint: None, s3_region: None, + wait: 0, }; save("round-trip", &config).expect("configuration should save"); @@ -381,4 +428,4 @@ s3_bucket = "example-bucket" fs::remove_dir_all(temp_home).ok(); } -} \ No newline at end of file +} diff --git a/src/file_sync/config.rs b/src/file_sync/config.rs new file mode 100644 index 0000000..902b0ff --- /dev/null +++ b/src/file_sync/config.rs @@ -0,0 +1,44 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Config { + pub admin_email: Option, + pub s3_bucket: String, + pub s3_client_id: String, + pub s3_client_secret: String, + pub s3_endpoint: Option, + pub s3_region: Option, + pub sfdx_auth_url: String, + pub wait: u64, +} + +impl Config { + pub(crate) fn s3_region(&self) -> &str { + self.s3_region.as_deref().unwrap_or("us-east-1") + } + + pub(crate) fn s3_bucket(&self) -> &str { + &self.s3_bucket + } + + pub(crate) fn s3_endpoint(&self) -> Option<&str> { + self.s3_endpoint.as_deref() + } + + pub(crate) fn s3_client_secret(&self) -> &str { + &self.s3_client_secret + } + + pub(crate) fn s3_client_id(&self) -> &str { + &self.s3_client_id + } + + pub(crate) fn sfdx_auth_url(&self) -> &str { + &self.sfdx_auth_url + } + + pub(crate) fn wait(&self) -> u64 { + self.wait + } +} + diff --git a/src/file_sync/document.rs b/src/file_sync/document.rs new file mode 100644 index 0000000..f4f68ef --- /dev/null +++ b/src/file_sync/document.rs @@ -0,0 +1,90 @@ +use anyhow::{Context, Result}; +use rustsf::{RestApi, SObjectOwner}; +use std::collections::HashMap; +use log::trace; +use crate::salesforce::models::{ContentDocumentLink, ContentVersion}; + +#[derive(Debug)] +pub(crate) struct Document { + content_version: ContentVersion, + content_document_links: Vec, +} +impl Document { + + pub(crate) fn content_version(&self) -> &ContentVersion { + &self.content_version + } + + pub(crate) fn from_content_versions_and_links( + content_versions: Vec, + content_document_links: Vec, + ) -> Vec { + let mut links_by_content_document_id: HashMap> = + HashMap::new(); + + for link in content_document_links { + if let Some(content_document_id) = link.content_document_id() { + links_by_content_document_id + .entry(content_document_id.to_string()) + .or_default() + .push(link); + } + } + + let mut result = Vec::new(); + for content_version in content_versions { + + let Some(content_document_id) = content_version.content_document_id() else { continue }; + + let Some(links) = + links_by_content_document_id.get(content_document_id) else { continue }; + + let mut content_document_links = Vec::new(); + for link in links { + if link.linked_entity_id() == content_version.first_publish_location_id().unwrap_or("None") + || link.linked_entity_id() == content_version.get_owner_id().unwrap_or("None") { continue } + + content_document_links.push(link.to_owned()); + } + + result.push( Document { + content_version, + content_document_links, + }) + } + result + } + + pub fn take(self) -> (ContentVersion, Vec) { + (self.content_version, self.content_document_links) + } + +} + +pub async fn new_batch(mut sf_client: &mut RestApi) -> Result> { + let records = ContentVersion::query_binary_records(&mut sf_client) + .await + .context("Failed to retrieve binary content docs")?; + + trace!("Query Response {} {:?}", records.len(), records); + + if records.is_empty() { + return Ok(Vec::new()); + } + + let content_document_ids= + records.iter() + .filter_map(|record| record.content_document_id()) + .collect(); + + let content_document_links = + ContentDocumentLink::query_by_content_document_ids(&mut sf_client, content_document_ids) + .await + .context("Failed to retrieve content document Links")?; + + trace!("Query Response {}, {:?}", content_document_links.len(), content_document_links); + + let result = Document::from_content_versions_and_links(records, content_document_links); + + Ok(result) +} diff --git a/src/file_sync/mod.rs b/src/file_sync/mod.rs index 67fcfb4..df07443 100644 --- a/src/file_sync/mod.rs +++ b/src/file_sync/mod.rs @@ -1,37 +1,238 @@ -pub fn run() { - unimplemented!("File sync not yet implemented"); +use anyhow::{Context, Result, anyhow}; +use bytes::Bytes; +use log::{error, info}; +use mime_type::{MimeFormat, MimeType}; +use pluralizer::pluralize; +use rustsf::{RestApi, SObject}; +use sha2::{Digest, Sha256}; - /* -let args = cli::Args::parse(); +pub(crate) mod config; +mod document; +mod validated_content_version; +use crate::file_sync::document::Document; +use crate::file_sync::validated_content_version::ValidatedContentVersion; +use crate::salesforce::models::{ContentDocumentLink, ContentVersion, ExternalDataSource}; +pub use config::Config as FileSyncConfig; - -let salesforce_client = salesforce::connect(&args.sfdx_auth_url).await?; -let s3_client = aws_s3::connect( - args.s3_client_id(), - args.s3_client_secret(), - args.s3_endpoint(), - args.s3_region(), -); - - -salesforce_client.query(r#"SELECT Id, Title, FileType, FileExtension, Description, -ContentLocation, Origin, VersionData, ContentDocumentId, FirstPublishLocationId, PathOnClient - FROM ContentVersion - WHERE IsMajorVersion = TRUE - AND ContentLocation = 'S' - LIMIT 1\ -"#) - - - -if terminator::CONFIG.is_fired() { - info!("Received termination signal, exiting..."); +async fn sleep(wait: u64) { + info!("Waiting {} seconds...", wait); + tokio::time::sleep(std::time::Duration::from_secs(wait)).await; // every 30 seconds } +pub async fn run(config: config::Config) -> Result<()> { + loop { + if crate::terminator::CONFIG.is_fired() { + info!("Received termination signal, exiting..."); + break; + } + match run_batch(&config).await { + Ok(size) if size > 0 => info!("Batch completed successfully, processed {} records", size), + Ok(_) => info!("No documents to process"), + Err(err) => { + error!("Failed to sync files: {:#}", err); + sleep(config.wait()).await; + continue; + } + } -info!("Done.");*/ - -} \ No newline at end of file + sleep(config.wait()).await; + } + + info!("Done..."); + + Ok(()) +} + +async fn run_batch(config: &config::Config) -> Result { + info!("Syncing files..."); + + let mut sf_client = crate::salesforce::connect(config.sfdx_auth_url()) + .await + .context("Failed to connect to Salesforce")?; + + let s3_client = crate::rustfs::initialise( + config.s3_endpoint(), + config.s3_region(), + config.s3_client_id(), + config.s3_client_secret(), + ) + .await; + + let eds_id = get_external_data_source_id(&mut sf_client).await + .context("ExternalDataSource query failed")?; + + let batch = document::new_batch(&mut sf_client) + .await + .context("Failed to retrieve documents")?; + + if batch.is_empty() { + return Ok(0); + } + + info!( + "Retrieved {}", + pluralize("binary content documents", batch.len() as isize, true) + ); + + let size = batch.len(); + for item in batch { + let content_document_id = item.content_version().content_document_id().unwrap_or("Unknown").to_string(); + info!("Processing record '{}'", content_document_id); + match process_record(&mut sf_client, &s3_client, config.s3_bucket(), &eds_id, item).await { + Ok(content_version) => info!( + "Record '{}' successfully converted to {}", + content_document_id, + content_version.content_document_id().unwrap_or("Unknown") + ), + Err(err) => { + error!("Failed to process ContentVersion record: {:#}", err); + // fixme - report failure to admin + continue; + } + } + } + + // Link new records + Ok(size) +} + +async fn get_external_data_source_id(sf_client: &mut RestApi) -> Result { + let query = "SELECT Id FROM ExternalDataSource WHERE DeveloperName = 'S3_EmailAttachmentStorage' WITH SYSTEM_MODE LIMIT 1"; + let eds_id = sf_client.query::(query) + .await + .context("Failed to query Salesforce")? + .records + .first() + .context("Could not file ExternalDataSource 'S3_EmailAttachmentStorage'")? + .id() + .context("Could not retrieve Id value from ExternalDataSource 'S3_EmailAttachmentStorage'")? + .to_owned(); + Ok(eds_id) +} + +async fn process_record( + sf_client: &mut RestApi, + s3_client: &aws_sdk_s3::Client, + bucket: &str, + eds_id: &str, + document: Document, +) -> Result { + let (content_version, content_document_link) = document.take(); + let content_version = + ValidatedContentVersion::new(content_version).context("Failed to validate ContentVersion record")?; + + info!( + "Retrieving binary blob data content ContentVersion record {}", + content_version.id() + ); + let version_data = fetch_content_data(sf_client, &content_version.version_data()) + .await + .context("Failed to fetch content version data")?; + + // Make sure download is not corrupt + if !has_valid_checksum(content_version.checksum(), &version_data) { + return Err(anyhow!("Checksum mismatch for ContentVersion record")); + } + + // Send file to S3 + let key = hex::encode(Sha256::digest(&version_data)); + info!( + "Sending file {} to S3 of {} KiB to S3 with key {}", + content_version.id(), + content_version.content_size() / 1024, + key + ); + + let ext = content_version.file_extension(); + print!("ext: {}\n", ext); + let content_type = match MimeType::from_ext(ext) { + Some(mime_type) => mime_type.to_string(), + None => "text/plain".to_string(), + }; + + crate::rustfs::command::create_object(&s3_client, &bucket, &key, &content_type, &version_data) + .await + .context("Failed to store file in S3")?; + + // todo - create records in Salesforce + let old_content_document_id = content_version.content_document_id().to_string(); + let new_content_version = content_version.to_content_version( + sf_client + .client() + .instance_url() + .unwrap_or("https://login.salesforce.com"), + &key, + &eds_id, + ); + + // Create new ContentVersion record + let new_content_version = sf_client + .create_sobject(new_content_version) + .await + .context("Failed to create new ContentVersion record")?; + + // Fetch ContentVersion.ContentDocumentId + let query = format!( + "SELECT ContentDocumentId FROM ContentVersion WHERE Id = '{}'", + new_content_version.id().unwrap_or("NULL") + ); + let new_content_document_id = sf_client + .query::(&query) + .await + .context("Failed to fetch ContentVersion.ContentDocumentId")? + .records + .first() + .context("Could not retrieve ContentVersion record, after it was created")? + .content_document_id() + .context("Failed to retrieve ContentDocumentId from ContentVersion record")? + .to_string(); + + let mut new_links = Vec::new(); + for old_link in content_document_link.iter() + { + new_links.push(ContentDocumentLink::new() + .set_content_document_id(new_content_document_id.to_string()) + .set_linked_entity_id(old_link.linked_entity_id().to_string()) + .set_share_type(old_link.share_type().to_string()) + .set_visibility(old_link.visibility().to_string()) + .to_owned() + ); + } + + sf_client.create(&mut new_links, true) + .await + .context("Failed to create new ContentDocumentLink records")?; + + // todo - remove old records from Salesforce + sf_client.delete_sobject("ContentDocument", &old_content_document_id) + .await + .context("Failed to delete old ContentDocumentLink records")?; + + Ok(new_content_version) +} + +async fn fetch_content_data(sf_client: &mut RestApi, version_data_url: &str) -> Result { + let url = format!( + "{}/{}", + sf_client + .client() + .instance_url() + .unwrap_or("https://login.salesforce.com"), + version_data_url + ); + + let response = sf_client.client().get(url.to_string(), vec![], vec![]).await?; + + if response.status().is_success() { + Ok(response.bytes().await?) + } else { + Err(anyhow!("API request failed: {}", response.status())) + } +} + +fn has_valid_checksum(checksum: &str, version_data: &[u8]) -> bool { + checksum == format!("{:?}", md5::compute(version_data)) +} diff --git a/src/file_sync/validated_content_version.rs b/src/file_sync/validated_content_version.rs new file mode 100644 index 0000000..efb926b --- /dev/null +++ b/src/file_sync/validated_content_version.rs @@ -0,0 +1,205 @@ +use crate::salesforce::models::ContentVersion; +use anyhow::{Context, Result}; +use rustsf::SObjectOwner; + +#[derive(Debug)] +pub struct ValidatedContentVersion { + checksum: String, + content_document_id: String, + content_size: u64, + file_extension: String, + file_type: String, + first_publish_location_id: String, + id: String, + owner_id: String, + origin: String, + path_on_client: String, + sharing_option: String, + sharing_privacy: String, + title: String, + version_data: String, + version_number: String, +} + +impl ValidatedContentVersion { + pub(crate) fn to_content_version(self, url: &str, key: &str, eds_id: &str) -> ContentVersion { + let url = format!("{}/lightning/cmp/c__caseAttachment?&c__key={}", url, key); + // let url = "https://wallpapercave.com/wp/wp2106696.jpg"; + ContentVersion::new() + .set_title(format!("{}.{}", self.title, self.file_extension).as_str()) + .set_content_url(&url) + .set_content_location("E") // S = Salesforce, E = External Data Source + .set_is_major_version(true) + .set_external_data_source_id(eds_id) + .set_external_document_info_1(&url) + .set_external_document_info_2(&self.path_on_client) + .set_origin("H") + .set_owner_id(Some(&self.owner_id)) + .set_first_publish_location_id(&self.first_publish_location_id) + .set_sharing_option(&self.sharing_option) + .set_sharing_privacy(&self.sharing_privacy) + .to_owned() + } +} + +impl ValidatedContentVersion { + pub(crate) fn checksum(&self) -> &str { + &self.checksum + } + + pub(crate) fn content_document_id(&self) -> &str { + &self.content_document_id + } + + pub(crate) fn content_size(&self) -> u64 { + self.content_size + } + + pub(crate) fn file_extension(&self) -> &str { + &self.file_extension + } + + pub(crate) fn id(&self) -> &str { + &self.id + } + + pub(crate) fn version_data(&self) -> &str { + &self.version_data + } +} + +impl ValidatedContentVersion { + pub fn new(mut content_version: ContentVersion) -> Result { + Ok(Self { + checksum: content_version + .take_checksum() + .context("Failed to retrieve content document ID")?, + content_document_id: content_version + .take_content_document_id() + .context("Failed to retrieve content document ID")?, + content_size: content_version + .take_content_size() + .context("Failed to retrieve content size")?, + file_extension: content_version + .take_file_extension() + .context("Failed to retrieve file extension")?, + file_type: content_version + .take_file_type() + .context("Failed to retrieve file type")?, + first_publish_location_id: content_version + .take_first_publish_location_id() + .context("Failed to retrieve first publish location ID")?, + id: content_version.take_id().context("Failed to retrieve record ID")?, + owner_id: content_version.take_owner_id().context("Failed to retrieve owner ID")?, + origin: content_version.take_origin().context("Failed to retrieve origin")?, + path_on_client: content_version + .take_path_on_client() + .context("Failed to retrieve path on client")?, + sharing_option: content_version + .take_sharing_option() + .context("Failed to retrieve sharing option")?, + sharing_privacy: content_version + .take_sharing_privacy() + .context("Failed to retrieve sharing privacy")?, + title: content_version.take_title().context("Failed to retrieve title")?, + version_data: content_version + .take_version_data() + .context("Failed to retrieve version data")?, + version_number: content_version + .take_version_number() + .context("Failed to retrieve version number")?, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn content_version(overrides: serde_json::Value) -> ContentVersion { + let mut value = json!({ + "attributes": { + "type": "ContentVersion", + "url": "", + }, + "Checksum": "5d41402abc4b2a76b9719d911017c592", + "ContentDocumentId": "069000000000001", + "ContentLocation": "S", + "ContentSize": 5, + "Description": null, + "FileExtension": "txt", + "FileType": "TEXT", + "FirstPublishLocationId": "001000000000001", + "IsDeleted": false, + "IsExternallyAvailable": false, + "IsLatest": true, + "IsMajorVersion": true, + "Origin": "H", + "PathOnClient": "hello.txt", + "ReasonForChange": null, + "SharingOption": "A", + "SharingPrivacy": "N", + "TagCsv": null, + "TextPreview": null, + "Title": "hello", + "VersionData": [104, 101, 108, 108, 111], + "VersionNumber": "1" + }); + + let object = value + .as_object_mut() + .expect("base ContentVersion JSON should be an object"); + let overrides = overrides + .as_object() + .expect("ContentVersion overrides should be an object"); + + for (key, override_value) in overrides { + object.insert(key.clone(), override_value.clone()); + } + + serde_json::from_value(value).expect("ContentVersion JSON should deserialize") + } + + #[test] + fn new_returns_validated_content_version_when_required_fields_and_checksum_are_valid() { + let content_version = content_version(json!({})); + + let validated = ValidatedContentVersion::new(&content_version).expect("valid ContentVersion should validate"); + + assert_eq!(validated.content_document_id, "069000000000001"); + assert_eq!(validated.file_extension, "txt"); + assert_eq!(validated.file_type, "TEXT"); + assert_eq!(validated.first_publish_location_id, "001000000000001"); + assert_eq!(validated.origin, "H"); + assert_eq!(validated.path_on_client, "hello.txt"); + assert_eq!(validated.sharing_option, "A"); + assert_eq!(validated.sharing_privacy, "N"); + assert_eq!(validated.title, "hello"); + assert_eq!(validated.version_data, b"hello"); + assert_eq!(validated.version_number, "1"); + } + + #[test] + fn new_fails_when_checksum_does_not_match_version_data() { + let content_version = content_version(json!({ + "Checksum": "00000000000000000000000000000000" + })); + + let error = + ValidatedContentVersion::new(&content_version).expect_err("checksum mismatch should fail validation"); + + assert!(error.to_string().contains("ContentVersion checksum mismatch")); + } + + #[test] + fn new_fails_when_required_field_is_missing() { + let content_version = content_version(json!({ + "Title": null + })); + + let error = ValidatedContentVersion::new(&content_version).expect_err("missing title should fail validation"); + + assert!(error.to_string().contains("Failed to retrieve title")); + } +} diff --git a/src/main.rs b/src/main.rs index 95dfd35..8416313 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,13 +1,12 @@ //! # IMoje CLI -//! +//! //! CLI features; //! - **file-sync** Syncs the contents of a Salesforce File to S3 -//! - **set** Sets the value of a configuration +//! - **set** Sets the value of a configuration //! - **list** Lists the contents of a configuration file` use std::sync::OnceLock; -mod aws_s3; mod cli; mod config; mod logger; @@ -15,11 +14,11 @@ mod salesforce; mod terminator; mod file_sync; - +mod rustfs; pub static APP_CONFIG: OnceLock = OnceLock::new(); -#[::tokio::main] -async fn main() { +fn main() { + cli::run(); } diff --git a/src/rustfs/command/mod.rs b/src/rustfs/command/mod.rs new file mode 100644 index 0000000..ae9604f --- /dev/null +++ b/src/rustfs/command/mod.rs @@ -0,0 +1,992 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +use anyhow::{Context, Result}; + +/// Creates an object in an Amazon S3 bucket. +/// +/// This function uploads an object to the specified S3 bucket with the provided key, content type, +/// and body. It uses the AWS SDK for Rust to perform the operation. The body of the object is +/// passed as a byte slice and converted into an S3-compatible byte stream during the process. +/// +/// # Parameters +/// +/// * `client` - A reference to an instance of the `aws_sdk_s3::Client` used to interact with S3. +/// * `bucket_name` - The name of the S3 bucket where the object will be created. +/// * `key` - The key (path) for the object in the bucket. +/// * `content_type` - The MIME type of the content being uploaded (e.g., `application/json`, `text/plain`). +/// * `body` - The content of the object as a byte slice. +/// +/// # Returns +/// +/// * `Ok(aws_sdk_s3::operation::put_object::PutObjectOutput)` - If the operation is successful, +/// returns the output containing the response from S3. +/// * `Err(S3Error)` - If the operation fails, returns an `S3Error` containing details about the error. +/// +/// # Errors +/// +/// This function propagates errors from the AWS SDK and wraps them in the custom `S3Error` type +/// to provide additional context. Possible errors include network issues, permissions issues, +/// invalid bucket or key names, or other S3-related issues. +/// +/// # Example +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use your_crate::S3Error; // Replace with the actual path to your S3Error type. +/// +/// #[tokio::main] +/// async fn main() { +/// let config = aws_config::load_from_env().await; +/// let client = Client::new(&config); +/// +/// let bucket_name = "example-bucket"; +/// let key = "example-key"; +/// let content_type = "text/plain"; +/// let body = b"Hello, S3!"; +/// +/// match create_object(&client, bucket_name, key, content_type, body) { +/// Ok(output) => { +/// println!("Object created successfully: {:?}", output); +/// } +/// Err(e) => { +/// eprintln!("Failed to create object: {:?}", e); +/// } +/// } +/// } +/// ``` +/// +/// # Notes +/// +/// * This function internally uses `block_on`, which blocks the current thread until +/// the asynchronous operation is completed. Avoid using it in asynchronous contexts +/// where full async support is required (e.g., in an asynchronous runtime like Tokio). +/// * Make sure the provided bucket and key meet S3 naming conventions and the required +/// permissions are granted for the specified operation. +pub async fn create_object( + client: &aws_sdk_s3::Client, + bucket_name: &str, + key: &str, + content_type: &str, + body: &[u8], +) -> Result { + let byte_stream = aws_sdk_s3::primitives::ByteStream::from(Vec::from(body)); + client + .put_object() + .bucket(bucket_name) + .key(key) + .content_type(content_type) + .body(byte_stream) + .send() + .await + .context("Failed to create object in S3") +} +/* +/// Copies an object from one bucket to another in an Amazon S3 service. +/// +/// # Arguments +/// +/// * `client` - An instance of the `aws_sdk_s3::Client` used to communicate with S3. +/// * `source_bucket` - The name of the source S3 bucket. +/// * `destination_bucket` - The name of the destination S3 bucket. +/// * `source_object` - The key (file name) of the object in the source bucket. +/// * `destination_object` - The key (file name) of the object in the destination bucket. +/// +/// # Returns +/// +/// * `Ok(())` if the object was successfully copied. +/// * `Err(S3Error)` if an error occurred during the copy operation. +/// +/// # Example +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use your_crate::copy_object; // Replace with your module path +/// +/// #[tokio::main] +/// async fn main() { +/// let client = Client::new(&aws_config::load_from_env().await); +/// let source_bucket = "source-bucket-name"; +/// let destination_bucket = "dest-bucket-name"; +/// let source_object = "source-key"; +/// let destination_object = "dest-key"; +/// +/// if let Err(err) = copy_object(&client, source_bucket, destination_bucket, source_object, destination_object).await { +/// eprintln!("Failed to copy object: {}", err); +/// } +/// } +/// ``` +/// +/// # Notes +/// +/// * The `source_bucket` and `source_object` are used to construct the `copy_source`, which follows the format: +/// `"/"` +/// * Ensure the AWS credentials used by the client have the necessary S3 permissions to read from the source bucket +/// and write to the destination bucket. +/// * The response will include the `ETag` of the copied object, if available. +#[allow(unused)] +pub async fn copy_object( + client: &aws_sdk_s3::Client, + source_bucket: &str, + destination_bucket: &str, + source_object: &str, + destination_object: &str, +) -> Result<()> { + let source_key = format!("{source_bucket}/{source_object}"); + let response = client + .copy_object() + .copy_source(&source_key) + .bucket(destination_bucket) + .key(destination_object) + .send() + .await + .context("Failed to copy object")?; + + println!( + "Copied from {source_key} to {destination_bucket}/{destination_object} with etag {}", + response + .copy_object_result + .unwrap_or_else(|| aws_sdk_s3::types::CopyObjectResult::builder().build()) + .e_tag() + .unwrap_or("missing") + ); + Ok(()) +} + +/// Asynchronously removes an object from an S3 bucket. +/// +/// # Arguments +/// +/// * `client` - An instance of `aws_sdk_s3::Client` used to interact with the S3 service. +/// * `bucket` - The name of the S3 bucket containing the object to be removed. +/// * `key` - The key (path) of the object to remove from the bucket. +/// +/// # Returns +/// +/// * `Result<(), S3Error>` - Returns `Ok(())` if the object was successfully removed. +/// Returns `Err(S3Error)` if an error occurred during the operation. +/// +/// # Errors +/// +/// This function returns an error of type `S3Error` if the underlying request to delete the +/// object fails. However, there are no modeled errors specific to deleting an object in +/// the AWS SDK. +/// +/// # Example +/// +/// ```rust +/// # async fn example() -> Result<(), S3Error> { +/// use aws_sdk_s3::Client; +/// +/// let client = Client::new(&aws_config::load_from_env().await); +/// let bucket = "example-bucket"; +/// let key = "example-key"; +/// +/// remove_object(&client, bucket, key).await?; +/// # Ok(()) +/// # } +/// ``` +#[allow(unused)] +pub async fn remove_object( + client: &aws_sdk_s3::Client, + bucket: &str, + key: &str, +) -> Result<()> { + client + .delete_object() + .bucket(bucket) + .key(key) + .send() + .await + .context("Failed to remove object")?; + + // There are no modeled errors to handle when deleting an object. + + Ok(()) +} + +/// Downloads an object from an S3 bucket using the provided S3 client. +/// +/// # Arguments +/// +/// * `client` - A reference to an `aws_sdk_s3::Client` instance used to interact with the AWS S3 service. +/// * `bucket_name` - The name of the S3 bucket where the object is stored. +/// * `key` - The key (path) of the object to be downloaded within the S3 bucket. +/// +/// # Returns +/// +/// This function returns a `Result`: +/// * `Ok(aws_sdk_s3::operation::get_object::GetObjectOutput)` - If the object is successfully downloaded, +/// it contains the output of the `get_object` operation, which includes the object's data and metadata. +/// * `Err(S3Error)` - If an error occurs during the operation, it returns an S3Error converted from the SDK error response. +/// +/// # Errors +/// +/// This function can fail if: +/// * The specified bucket or key does not exist or the user does not have sufficient permissions. +/// * There are network issues or problems communicating with the AWS S3 service. +/// * Any issue occurs during the `get_object` operation, such as invalid credentials. +/// +/// # Example +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use my_crate::download_object; // Replace `my_crate` with the actual module path +/// +/// #[tokio::main] +/// async fn main() { +/// let client = Client::new(&aws_config::load_from_env().await); +/// let bucket_name = "my-bucket"; +/// let key = "path/to/my/object.txt"; +/// +/// match download_object(&client, bucket_name, key).await { +/// Ok(output) => { +/// println!("Object downloaded successfully!"); +/// // Use output.body to access the downloaded content. +/// } +/// Err(err) => { +/// eprintln!("Failed to download object: {}", err); +/// } +/// } +/// } +/// ``` +/// +/// # Dependencies +/// +/// * This function requires the `aws-sdk-s3` crate and its associated types (`Client`, `GetObjectOutput`). +/// * The `S3Error` type should be defined in your project to handle errors appropriately. +#[allow(unused)] +pub async fn download_object( + client: &aws_sdk_s3::Client, + bucket_name: &str, + key: &str, +) -> Result { + client + .get_object() + .bucket(bucket_name) + .key(key) + .send() + .await + .map_err(S3Error::from) +} + +/// Asynchronously uploads an object to an S3 bucket using the given AWS SDK S3 client. +/// +/// # Arguments +/// +/// * `client` - A reference to an `aws_sdk_s3::Client` instance used to interact with the S3 service. +/// * `bucket_name` - The name of the bucket where the object will be uploaded. +/// * `file_name` - The path to the local file that will be uploaded as the object. +/// * `key` - The key (path) under which the object will be stored in the bucket. +/// +/// # Returns +/// +/// Returns a `Result` containing the `PutObjectOutput` (on success) or an `S3Error` (on failure). +/// +/// # Errors +/// +/// This function will return an error if: +/// * The provided `file_name` cannot be read (e.g., file doesn't exist or lacks permissions). +/// * The `put_object` API call to S3 fails due to network issues, invalid credentials, or other AWS errors. +/// +/// # Examples +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use my_crate::upload_object; +/// +/// #[tokio::main] +/// async fn main() { +/// let client = Client::new(&aws_config::load_from_env().await); +/// let bucket_name = "example-bucket"; +/// let file_name = "path/to/file.txt"; +/// let key = "uploaded-file.txt"; +/// +/// match upload_object(&client, bucket_name, file_name, key).await { +/// Ok(output) => println!("File uploaded successfully: {:?}", output), +/// Err(e) => eprintln!("Error uploading file: {:?}", e), +/// } +/// } +/// ``` +#[allow(unused)] +pub async fn upload_object( + client: &aws_sdk_s3::Client, + bucket_name: &str, + file_name: &str, + key: &str, +) -> Result { + let body = aws_sdk_s3::primitives::ByteStream::from_path(std::path::Path::new(file_name)).await; + client + .put_object() + .bucket(bucket_name) + .key(key) + .body(body.unwrap()) + .send() + .await + .map_err(S3Error::from) +} + +/// Lists objects from an S3 bucket using the AWS SDK for Rust. +/// +/// This function retrieves objects from the specified S3 bucket in batches of 10, +/// printing the key of each object to the standard output. If an error occurs during +/// pagination, the error is logged using the standard error output. +/// +/// # Parameters +/// - `client`: A reference to an `aws_sdk_s3::Client` instance used to interact with Amazon S3. +/// - `bucket`: The name of the bucket from which to list objects. +/// +/// # Returns +/// - `Ok(())` if the operation completes successfully. +/// - `Err(S3Error)` if an error occurs while listing objects. +/// +/// # Errors +/// This function will return an `Err(S3Error)` if there is an issue with the S3 client +/// or the request itself. +/// +/// # Examples +/// ``` +/// use aws_sdk_s3::Client; +/// use my_crate::list_objects; // Replace `my_crate` with the actual crate name. +/// +/// #[tokio::main] +/// async fn main() { +/// let client = Client::new(&aws_config::load_from_env().await); +/// let bucket_name = "example-bucket"; +/// +/// if let Err(err) = list_objects(&client, bucket_name) { +/// eprintln!("Error listing objects: {:?}", err); +/// } +/// } +/// ``` +/// +/// # Notes +/// - The function uses `block_on` to await the asynchronous paginator responses. In a +/// production system, consider restructuring to avoid blocking in asynchronous contexts. +/// - The maximum number of keys retrieved per batch is 10, as defined by the `max_keys` +/// parameter. +/// +/// # Dependencies +/// - The AWS SDK for Rust: `aws-sdk-s3` +/// - A Tokio runtime to properly execute the asynchronous code (external in the calling context). +#[allow(unused)] +pub fn list_objects(client: &aws_sdk_s3::Client, bucket: &str) -> Result<(), S3Error> { + let mut response = client + .list_objects_v2() + .bucket(bucket.to_owned()) + .max_keys(10) // In this example, go 10 at a time. + .into_paginator() + .send(); + + while let Some(result) = block_on(response.next()) { + match result { + Ok(output) => { + for object in output.contents() { + println!(" - {}", object.key().unwrap_or("Unknown")); + } + } + Err(err) => { + eprintln!("{err:?}") + } + } + } + + Ok(()) +} + +/// Lists all the S3 buckets in the current AWS account using the provided S3 client. +/// +/// # Arguments +/// +/// * `client` - A reference to an instance of `aws_sdk_s3::Client`, used to communicate with the AWS S3 service. +/// +/// # Returns +/// +/// * `Ok(())` if the buckets are successfully listed and printed to the standard output. +/// * `Err(S3Error)` if an error occurs while attempting to list the buckets. +/// +/// # Behavior +/// +/// This function retrieves the list of S3 buckets associated with the AWS account linked to the given `client`. +/// It prints the total number of buckets and the names of each bucket to the standard output. If an error occurs +/// during the process, it logs the error message to the standard output and returns an appropriate error. +/// +/// # Example +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use your_crate::list_buckets; // Replace `your_crate` with the actual crate name containing this function. +/// +/// #[tokio::main] +/// async fn main() -> Result<(), Box> { +/// let config = aws_config::load_from_env().await; +/// let client = Client::new(&config); +/// +/// list_buckets(&client)?; +/// +/// Ok(()) +/// } +/// ``` +/// +/// # Notes +/// +/// * This function uses `block_on` for synchronous execution of the async function `client.list_buckets().send()`. +/// It is generally encouraged to utilize asynchronous programming rather than blocking, especially in applications +/// that are designed to be async. +/// * Make sure your AWS credentials and permissions are properly configured to call the `list_buckets` operation. +/// +/// # Errors +/// +/// This function may return an `Err(S3Error)` in the following cases: +/// * AWS service issues, such as network disruptions or a permissions problem. +/// * Invalid AWS authentication or expired credentials. +#[allow(unused)] +pub fn list_buckets(client: &aws_sdk_s3::Client) -> Result<(), S3Error> { + match block_on(client.list_buckets().send()) { + Ok(res) => { + println!("Total buckets number is {:?}", res.buckets().len()); + for bucket in res.buckets() { + println!("Bucket: {:?}", bucket.name()); + } + } + Err(e) => { + println!("Error listing buckets: {:?}", e); + } + } + + Ok(()) +} + +/// Asynchronously clears all objects from the specified Amazon S3 bucket. +/// +/// # Arguments +/// +/// * `client` - An instance of `aws_sdk_s3::Client` used to interact with S3. +/// * `bucket_name` - The name of the bucket to clear. +/// +/// # Returns +/// +/// On success, returns a `Result` wrapping a `Vec` containing the keys of the deleted objects. +/// If there are still objects remaining in the bucket after attempting to clear it, +/// an `S3Error` is returned. +/// +/// # Errors +/// +/// Returns an `S3Error` if: +/// * There is an issue communicating with the S3 service (e.g., network issues). +/// * The bucket contains objects that were not successfully deleted. +/// * Any other AWS S3 API-related errors occur. +/// +/// # Example +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use my_crate::clear_bucket; +/// +/// #[tokio::main] +/// async fn main() -> Result<(), S3Error> { +/// let config = aws_config::load_from_env().await; +/// let client = Client::new(&config); +/// let bucket_name = "my_bucket"; +/// +/// match clear_bucket(&client, bucket_name).await { +/// Ok(deleted_keys) => { +/// println!("Deleted objects: {:?}", deleted_keys); +/// } +/// Err(e) => { +/// eprintln!("Error clearing bucket: {:?}", e); +/// } +/// } +/// Ok(()) +/// } +/// ``` +/// +/// # Notes +/// +/// This function performs the following steps: +/// 1. Lists the objects currently in the specified bucket. +/// 2. Deletes all listed objects. +/// 3. Verifies the bucket is empty by listing objects again. +/// +/// If the bucket is not empty after attempting deletion, an error is returned. +/// +/// Ensure that you have the necessary permissions for the S3 bucket (e.g., `s3:ListBucket` and `s3:DeleteObject`). +#[allow(unused)] +pub async fn clear_bucket( + client: &aws_sdk_s3::Client, + bucket_name: &str, +) -> Result, S3Error> { + let objects = client.list_objects_v2().bucket(bucket_name).send().await?; + + // delete_objects no longer needs to be mutable. + let objects_to_delete: Vec = objects + .contents() + .iter() + .filter_map(|obj| obj.key()) + .map(String::from) + .collect(); + + if objects_to_delete.is_empty() { + return Ok(vec![]); + } + + let return_keys = objects_to_delete.clone(); + + delete_objects(client, bucket_name, objects_to_delete).await?; + + let objects = client.list_objects_v2().bucket(bucket_name).send().await?; + + eprintln!("{objects:?}"); + + match objects.key_count { + Some(0) => Ok(return_keys), + _ => Err(S3Error::/**/new("There were still objects left in the bucket.")), + } +} + +/// Asynchronously deletes a list of objects from an S3 bucket using the AWS SDK for Rust. +/// +/// # Arguments +/// +/// * `client` - A reference to an instance of `aws_sdk_s3::Client` used to interact with the S3 service. +/// * `bucket_name` - A string slice specifying the name of the bucket from which the objects will be deleted. +/// * `objects_to_delete` - A vector containing the keys (file names) of the objects to delete from the bucket. +/// +/// # Returns +/// +/// * `Result<(), S3Error>` - Returns `Ok(())` if all objects are successfully deleted. +/// Returns an `Err(S3Error)` in case of any errors during the process. +/// +/// # Errors +/// +/// This function will return an error in the following cases: +/// - If there is any issue building an `ObjectIdentifier` for the objects to be deleted. +/// - If there is an error constructing the `Delete` request payload. +/// - If the S3 `delete_objects` API call fails for any reason. +/// +/// # Example +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use std::error::Error; +/// +/// #[tokio::main] +/// async fn main() -> Result<(), Box> { +/// let client = Client::new(&aws_sdk_s3::config::Config::builder().build()?); +/// let bucket_name = "example-bucket"; +/// let objects_to_delete = vec!["file1.txt".to_string(), "file2.txt".to_string()]; +/// +/// delete_objects(&client, bucket_name, objects_to_delete).await?; +/// println!("Objects deleted successfully."); +/// +/// Ok(()) +/// } +/// ``` +/// +/// # Notes +/// +/// - The function requires the `client` to be properly configured with AWS credentials and region. +/// - Ensure that the calling IAM role/user has appropriate permissions to delete objects from the specified bucket. +/// +#[allow(unused)] +pub async fn delete_objects( + client: &aws_sdk_s3::Client, + bucket_name: &str, + objects_to_delete: Vec, +) -> Result<(), S3Error> { + // Push into a mut vector to use `?` early return errors while building object keys. + let mut delete_object_ids: Vec = vec![]; + for obj in objects_to_delete { + let obj_id = aws_sdk_s3::types::ObjectIdentifier::builder() + .key(obj) + .build() + .map_err(|err| { + S3Error::new(format!("Failed to build key for delete_object: {err:?}")) + })?; + delete_object_ids.push(obj_id); + } + + client + .delete_objects() + .bucket(bucket_name) + .delete( + aws_sdk_s3::types::Delete::builder() + .set_objects(Some(delete_object_ids)) + .build() + .map_err(|err| { + S3Error::new(format!("Failed to build delete_object input {err:?}")) + })?, + ) + .send() + .await?; + Ok(()) +} + +/// Asynchronously creates an S3 bucket in the specified AWS region. +/// +/// # Arguments +/// +/// * `client` - A reference to an `aws_sdk_s3::Client` instance used for S3 operations. +/// * `bucket_name` - The name of the bucket to be created. +/// * `region` - A reference to the AWS region where the bucket should be created. +/// +/// # Returns +/// +/// This function returns a `Result`: +/// * On success: +/// - `Ok(Some(CreateBucketOutput))` if the bucket was successfully created. +/// - `Ok(None)` if the bucket already exists or is already owned by the user. +/// * On failure: +/// - `Err(S3Error)` if there was an error other than the bucket already existing or being owned. +/// +/// # Errors +/// +/// Returns an `S3Error` if the bucket creation fails due to reasons other than: +/// * The bucket already exists. +/// * The bucket is already owned by the user. +/// +/// # Example +/// +/// ```rust +/// use aws_sdk_s3::Client; +/// use aws_config::Region; +/// use crate::S3Error; +/// +/// #[tokio::main] +/// async fn main() -> Result<(), S3Error> { +/// let config = aws_config::load_from_env().await; +/// let client = Client::new(&config); +/// let region = Region::new("us-east-1"); +/// +/// let result = create_bucket(&client, "my-new-bucket", ®ion).await?; +/// match result { +/// Some(output) => println!("Bucket created successfully: {:?}", output), +/// None => println!("Bucket already exists or is owned by you."), +/// } +/// +/// Ok(()) +/// } +/// ``` +#[allow(unused)] +pub async fn create_bucket( + client: &aws_sdk_s3::Client, + bucket_name: &str, + region: &aws_config::Region, +) -> Result, S3Error> { + let constraint = aws_sdk_s3::types::BucketLocationConstraint::from(region.to_string().as_str()); + let cfg = aws_sdk_s3::types::CreateBucketConfiguration::builder() + .location_constraint(constraint) + .build(); + let create = client + .create_bucket() + .create_bucket_configuration(cfg) + .bucket(bucket_name) + .send() + .await; + + // BucketAlreadyExists and BucketAlreadyOwnedByYou are not problems for this task. + create.map(Some).or_else(|err| { + if err + .as_service_error() + .map(|se| se.is_bucket_already_exists() || se.is_bucket_already_owned_by_you()) + == Some(true) + { + Ok(None) + } else { + Err(S3Error::from(err)) + } + }) +} + +/// Deletes an S3 bucket with the specified name using the provided AWS S3 client. +/// +/// # Parameters +/// - `client`: An instance of [`aws_sdk_s3::Client`] to interact with the S3 service. +/// - `bucket_name`: A string slice that holds the name of the bucket to be deleted. +/// +/// # Returns +/// - `Ok(())`: If the bucket is successfully deleted or if the bucket does not exist. +/// - `Err(S3Error)`: If an error occurs during the operation and it is not a `NoSuchBucket` error. +/// +/// # Behavior +/// - The function makes an asynchronous request to delete the specified bucket. +/// - If the bucket does not exist (indicated by the `NoSuchBucket` error), the function will still return `Ok(())`. +/// - For any other errors, the function returns an `Err(S3Error)` containing the original error. +/// +/// # Example +/// ```rust +/// use aws_sdk_s3::Client; +/// +/// #[tokio::main] +/// async fn main() { +/// let config = aws_config::load_from_env().await; +/// let client = Client::new(&config); +/// +/// match delete_bucket(&client, "example-bucket-name").await { +/// Ok(()) => println!("Bucket deleted successfully!"), +/// Err(err) => eprintln!("Failed to delete bucket: {:?}", err), +/// } +/// } +/// ``` +/// +/// # Errors +/// This function wraps any errors from the `aws_sdk_s3` crate that are not +/// specifically `NoSuchBucket`, converting them into an `S3Error`. +/// +/// # Notes +/// Ensure that the AWS credentials and permissions allow deleting the bucket, +/// and that the bucket is empty before attempting to delete it. +#[allow(unused)] +pub async fn delete_bucket(client: &aws_sdk_s3::Client, bucket_name: &str) -> Result<(), S3Error> { + let resp = client.delete_bucket().bucket(bucket_name).send().await; + match resp { + Ok(_) => Ok(()), + Err(err) => { + if err + .as_service_error() + .and_then(aws_sdk_s3::error::ProvideErrorMetadata::code) + == Some("NoSuchBucket") + { + Ok(()) + } else { + Err(S3Error::from(err)) + } + } + } +} + + +#[cfg(test)] +mod test { + use std::env::temp_dir; + + use aws_config::Region; + use aws_smithy_runtime::client::http::test_util::StaticReplayClient; + use sdk_examples_test_utils::{client_config, single_shot_client, test_event}; + use tokio::{fs::File, io::AsyncWriteExt}; + use uuid::Uuid; + + use crate::{ + clear_bucket, copy_object, create_bucket, delete_bucket, download_object, list_objects, + upload_object, + }; + + #[tokio::test] + async fn test_delete_bucket() { + let client = single_shot_client!( + sdk: aws_sdk_s3, + status: 200, + response: r#""# + ); + + let resp = delete_bucket(&client, "bucket_name").await; + + assert!(resp.is_ok(), "{resp:?}"); + } + + #[tokio::test] + async fn test_delete_missing_bucket() { + let client = single_shot_client!( + sdk: aws_sdk_s3, + status: 404, + response: r#"NoSuchBucketThe specified bucket does not existbucket_nameREQUESTHOSTID="# + ); + + let resp = delete_bucket(&client, "bucket_name").await; + + assert!(resp.is_ok(), "{resp:?}"); + } + + #[tokio::test] + async fn test_delete_objects() { + let client = aws_sdk_s3::Client::from_conf( + client_config!(aws_sdk_s3) + .http_client(StaticReplayClient::new(vec![ + // client.list_objects_v2().bucket(bucket_name) + test_event!( + r#""#, + ( + 200, + r#" + test + obj1 + obj2 + 2 + "# + ) + ), + // client.delete_objects().delete(...(delete_objects)...)) + test_event!(r#""#, (200, r#" + + + true + obj1 + + + true + obj2 + + + "#)), + // client.list_objects_v2().bucket(bucket_name) + test_event!( + r#""#, + ( + 200, + r#"test + 0 + "# + ) + ), + ])) + .build(), + ); + + let resp = clear_bucket(&client, "bucket_name").await; + + assert!(resp.is_ok(), "{resp:?}"); + } + + #[tokio::test] + async fn test_delete_objects_failed() { + let client = aws_sdk_s3::Client::from_conf( + client_config!(aws_sdk_s3) + .http_client(StaticReplayClient::new(vec![ + // client.list_objects_v2().bucket(bucket_name) + test_event!( + r#""#, + ( + 200, + r#" + test + obj1 + obj2 + 2 + "# + ) + ), + // client.delete_objects().delete(...(delete_objects)...)) + test_event!( + r#""#, + ( + 200, + r#" + + + true + obj1 + + + true + obj2 + + + "# + ) + ), + // client.list_objects_v2().bucket(bucket_name) + test_event!( + r#""#, + ( + 200, + r#" + test + obj3 + 1 + "# + ) + ), + ])) + .build(), + ); + + let resp = clear_bucket(&client, "bucket_name").await; + + assert!(resp.is_err(), "{resp:?}"); + } + + #[tokio::test] + async fn test_list_objects() { + let client = single_shot_client!( + sdk: aws_sdk_s3, + status: 200, + response: r#" + + test +"# + ); + + let resp = list_objects(&client, "bucket_name").await; + assert!(resp.is_ok(), "{resp:?}"); + } + + #[tokio::test] + async fn test_copy_object() { + let client = single_shot_client!( + sdk: aws_sdk_s3, + status: 200, + response: r#""# + ); + + let bucket = "bucket_name"; + let resp = copy_object(&client, bucket, bucket, "object_key", "target_key").await; + assert!(resp.is_ok(), "{resp:?}"); + } + + #[tokio::test] + async fn test_download_object() { + let client = single_shot_client!( + sdk: aws_sdk_s3, + status: 200, + response: r#""# + ); + + let resp = download_object(&client, "bucket_name", "key").await; + assert!(resp.is_ok(), "{resp:?}"); + } + + #[tokio::test] + async fn test_upload_object() { + let client = single_shot_client!( + sdk: aws_sdk_s3, + status: 200, + response: r#""# + ); + + let file_name = { + let mut dir = temp_dir(); + let file_name = format!("{}.txt", Uuid::new_v4()); + dir.push(file_name); + let file_name = dir.clone(); + let file_name = file_name.to_str().unwrap().to_string(); + + let mut file = File::create(dir).await.unwrap(); + let written_bytes = file.write("test file".as_bytes()).await.unwrap(); + + println!("Wrote bytes to s3 {written_bytes}"); + + file_name + }; + + let resp = upload_object(&client, "bucket_name", file_name.as_str(), "key").await; + + assert!(resp.is_ok(), "{resp:?}"); + } + + #[tokio::test] + async fn test_create_bucket() { + let client = single_shot_client!( + sdk: aws_sdk_s3, + status: 200, + headers: vec![("Location", "test_location")], + response: r#""# + ); + + let resp = create_bucket(&client, "bucket_name", &Region::from_static("us-esst-1")).await; + assert!(resp.is_ok(), "{resp:?}"); + let output = resp.unwrap(); + assert!(output.is_some()); + assert_eq!(output.unwrap().location(), Some("test_location")); + } +} +*/ diff --git a/src/rustfs/mod.rs b/src/rustfs/mod.rs new file mode 100644 index 0000000..3443687 --- /dev/null +++ b/src/rustfs/mod.rs @@ -0,0 +1,67 @@ +//! Module that defines functionality for interacting with an S3-compatible storage service. +//! +use aws_config::Region; +use log::{debug, trace}; + +use aws_credential_types::provider::SharedCredentialsProvider; +pub mod command; + +/// Initializes and returns an AWS S3 client with the provided credentials and configuration. +/// +/// This function creates an S3 client using static credentials and supports both standard AWS endpoints +/// and custom S3-compatible endpoints (like RustFS/MinIO). The configuration is loaded synchronously +/// using `block_on`. +/// +/// # Parameters +/// +/// * `endpoint` - Optional custom S3 endpoint URL. If `None`, uses the default AWS S3 endpoint for the specified region. +/// * `region` - AWS region identifier (e.g., "us-east-1"). Used for both AWS and custom endpoints. +/// * `key` - AWS Access Key ID for authentication. +/// * `secret` - AWS Secret Access Key for authentication. +/// +/// # Returns +/// +/// Returns a configured `aws_sdk_s3::Client` ready for S3 operations. +/// +/// # Example +/// +/// ```rust +/// // With custom endpoint (e.g., RustFS) +/// let client = initialise( +/// &Some("https://s3.example.com".to_string()), +/// "us-east-1", +/// "access_key", +/// "secret_key" +/// ); +/// +/// // With default AWS endpoint +/// let client = initialise(&None, "us-east-1", "access_key", "secret_key"); +/// ``` +pub async fn initialise(endpoint: Option<&str>, region: &str, key: &str, secret: &str) -> aws_sdk_s3::Client { + debug!("Initialising S3 connection"); + + let creds = SharedCredentialsProvider::new(aws_credential_types::Credentials::new( + key, secret, None, None, "Static", + )); + + let config = match endpoint { + Some(endpoint) => { + trace!("S3 custom endpoint: {}", endpoint); + aws_config::from_env() + .region(Region::new(region.to_string())) + .endpoint_url(endpoint) + .credentials_provider(creds) + .load() + .await + } + None => { + aws_config::from_env() + .region(Region::new(region.to_string())) + .credentials_provider(creds) + .load() + .await + } + }; + + aws_sdk_s3::Client::new(&config) +} diff --git a/src/salesforce/mod.rs b/src/salesforce/mod.rs index d2050ed..34ebf92 100644 --- a/src/salesforce/mod.rs +++ b/src/salesforce/mod.rs @@ -12,4 +12,24 @@ pub async fn connect(sfdx_auth_url: &str) -> Result { .context("Unable to connect to Salesforce")?; Ok(RestApi::new(client)) -} \ No newline at end of file +} + +/* +pub(crate) fn retrieve_binary_content_docs(api: RestApi) -> Vec { +/* + match models::ContentVersion::query_binary_records(api) { + Ok(records) => records, + Err(e) => { + + println!("Error: {}", e); + Vec::new() + } + } + + let fields = ["Id", "Title", "FileType", "FileExtension", "Description", + "ContentLocation", "Origin", "VersionData", "ContentDocumentId", "FirstPublishLocationId", + "PathOnClient"]; + + api.query(format!("SELECT {} FROM ContentVersion"#).await.unwrap()*/ + +}*/ \ No newline at end of file diff --git a/src/salesforce/models.rs b/src/salesforce/models.rs deleted file mode 100644 index d43450c..0000000 --- a/src/salesforce/models.rs +++ /dev/null @@ -1,154 +0,0 @@ -use rustsf::DefSObject; - -#[DefSObject(sobject_type = "ContentVersion", fields="owner,audit,system")] -pub(crate) struct ContentVersion { - - /// The ID of the ContentDocument associated with this version. - #[serde(rename = "ContentDocumentId")] - content_document_id: Option, - - /// The location where the content is stored. Valid values include Salesforce, external, or linked content locations. - #[serde(rename = "ContentLocation")] - content_location: Option, - - /// The date and time when the file content was last modified. - #[serde(rename = "ContentModifiedDate")] - content_modified_date: Option, - - /// The ID of the user who last modified the file content. - #[serde(rename = "ContentModifiedById")] - content_modified_by_id: Option, - - /// The URL of the content when the file is stored outside Salesforce. - #[serde(rename = "ContentUrl")] - content_url: Option, - - /// The version number of the file. - #[serde(rename = "VersionNumber")] - version_number: Option, - - /// The title of the file. - #[serde(rename = "Title")] - title: Option, - - /// The description of the file. - #[serde(rename = "Description")] - description: Option, - - /// The reason this version of the file was created or changed. - #[serde(rename = "ReasonForChange")] - reason_for_change: Option, - - /// Controls whether users can share the file with other users, groups, or records. - #[serde(rename = "SharingOption")] - sharing_option: Option, - - /// Controls file privacy on records. - #[serde(rename = "SharingPrivacy")] - sharing_privacy: Option, - - /// The complete path of the file on the client machine when it was uploaded. - #[serde(rename = "PathOnClient")] - path_on_client: Option, - - /// Indicates whether this version is the latest version of the file. - #[serde(rename = "IsLatest")] - is_latest: bool, - - /// Indicates whether this version is a major version. - #[serde(rename = "IsMajorVersion")] - is_major_version: bool, - - /// Indicates whether the file can be used as an asset file. - #[serde(rename = "IsAssetEnabled")] - is_asset_enabled: bool, - - /// Indicates whether the file is managed by an external client. - #[serde(rename = "IsClientManaged")] - is_client_managed: bool, - - /// Indicates whether the file is available externally. - #[serde(rename = "IsExternallyAvailable")] - is_externally_available: bool, - - /// The ID of the external data source for externally stored content. - #[serde(rename = "ExternalDataSourceId")] - external_data_source_id: Option, - - /// External document information used to identify externally stored content. - #[serde(rename = "ExternalDocumentInfo1")] - external_document_info1: Option, - - /// Additional external document information used to identify externally stored content. - #[serde(rename = "ExternalDocumentInfo2")] - external_document_info2: Option, - - /// The boost value used to promote featured content. - #[serde(rename = "FeaturedContentBoost")] - featured_content_boost: Option, - - /// The date when the file was marked as featured content. - #[serde(rename = "FeaturedContentDate")] - featured_content_date: Option, - - /// The file extension of the uploaded file. - #[serde(rename = "FileExtension")] - file_extension: Option, - - /// The type of the file, such as PDF, PNG, WORD, or LINK. - #[serde(rename = "FileType")] - file_type: Option, - - /// The ID of the first location where the file was published. - #[serde(rename = "FirstPublishLocationId")] - first_publish_location_id: Option, - - /// The source from which the content originated. - #[serde(rename = "Origin")] - origin: Option, - - /// The ID of the Experience Cloud site or network associated with the file. - #[serde(rename = "NetworkId")] - network_id: Option, - - /// The number of positive ratings submitted for this file version. - #[serde(rename = "PositiveRatingCount")] - positive_rating_count: Option, - - /// The publishing status of the file version. - #[serde(rename = "PublishStatus")] - publish_status: Option, - - /// A comma-separated list of tags associated with the file. - #[serde(rename = "TagCsv")] - tag_csv: Option, - - /// A text preview of the file content. - #[serde(rename = "TextPreview")] - text_preview: Option, - - /// The body of the file, represented as base64-encoded content when retrieved through the API. - #[serde(rename = "VersionData")] - version_data: Option, - - /// The size of the file in bytes. - #[serde(rename = "ContentSize")] - content_size: Option, - - /// MD5 checksum for the file. - #[serde(rename = "Checksum")] - checksum: Option, - - /// The ID of the ContentBody record that stores the binary file data. - #[serde(rename = "ContentBodyId")] - content_body_id: Option, - - /// The ID of the user who created the file version. - #[serde(rename = "CreatedById")] - created_by_id: String, - - /// The date and time when the file version was created. - #[serde(rename = "CreatedDate")] - created_date: String, - -} \ No newline at end of file diff --git a/src/salesforce/models/content_document_link.rs b/src/salesforce/models/content_document_link.rs new file mode 100644 index 0000000..ef14e8d --- /dev/null +++ b/src/salesforce/models/content_document_link.rs @@ -0,0 +1,88 @@ +use rustsf::{DefSObject, RestApi}; + +#[DefSObject(sobject_type = "ContentDocumentLink")] +pub(crate) struct ContentDocumentLink { + /// The related Content Document Id + #[serde(rename = "ContentDocumentId")] + content_document_id: String, + + /// The related Entity Id + #[serde(rename = "LinkedEntityId")] + linked_entity_id: String, + + /// The type of sharing, V = Viewer permissions + #[serde(rename = "ShareType")] + share_type: String, + + /// The visibility, e.g., AllUsers + #[serde(rename = "Visibility")] + visibility: String, +} + +impl ContentDocumentLink { + pub fn content_document_id(&self) -> Option<&str> { + Some(&self.content_document_id) + } + + pub fn linked_entity_id(&self) -> &str { + &self.linked_entity_id + } + + pub fn query_fields() -> Vec<&'static str> { + vec!["ContentDocumentId", "Id", "LinkedEntityId", "ShareType", "Visibility"] + } + + pub async fn query_by_content_document_ids( + api: &mut RestApi, + content_document_ids: Vec<&str>, + ) -> anyhow::Result> { + let query = format!( + "SELECT {} FROM {} WHERE {} LIMIT 50", + Self::query_fields().join(","), + Self::sobject_type(), + format!( + "ContentDocumentId IN ({})", + content_document_ids + .into_iter() + .map(|value| format!("'{}'", value)) + .collect::>() + .join(",") + ) + ); + let result = Ok(api.query::(&query).await?.records); + result + } + + + pub fn share_type(&self) -> &str { + &self.share_type + } + + pub fn sobject_type() -> &'static str { + "ContentDocumentLink" + } + + pub fn set_content_document_id(&mut self, value: String) -> &mut Self { + self.content_document_id = value; + self + } + + pub fn set_visibility(&mut self, value: String) -> &mut Self { + self.visibility = value; + self + } + + pub fn set_linked_entity_id(&mut self, value: String) -> &mut Self { + self.linked_entity_id = value; + self + } + + pub fn set_share_type(&mut self, value: String) -> &mut Self { + self.share_type = value; + self + } + + pub fn visibility(&self) -> &str { + &self.visibility + } +} diff --git a/src/salesforce/models/content_version.rs b/src/salesforce/models/content_version.rs new file mode 100644 index 0000000..24efd96 --- /dev/null +++ b/src/salesforce/models/content_version.rs @@ -0,0 +1,270 @@ +use anyhow::Result; +use rustsf::{DefSObject, RestApi}; + +#[DefSObject(sobject_type = "ContentVersion", fields = "owner")] +pub(crate) struct ContentVersion { + /// MD5 checksum for the file. + #[serde(rename = "Checksum", skip_serializing_if = "Option::is_none")] + checksum: Option, + + /// The ID of the ContentDocument associated with this version. + #[serde(rename = "ContentDocumentId", skip_serializing_if = "Option::is_none")] + content_document_id: Option, + + /// The location where the content is stored. Valid values include Salesforce, external, or linked content locations. + #[serde(rename = "ContentLocation", skip_serializing_if = "Option::is_none")] + content_location: Option, + + /// The size of the file in bytes. + #[serde(rename = "ContentUrl", skip_serializing_if = "Option::is_none")] + content_url: Option, + + /// The size of the file in bytes. + #[serde(rename = "ContentSize", skip_serializing_if = "Option::is_none")] + content_size: Option, + + /// The description of the file. + #[serde(rename = "Description", skip_serializing_if = "Option::is_none")] + description: Option, + + #[serde(rename = "ExternalDataSourceId", skip_serializing_if = "Option::is_none")] + external_data_source_id: Option, + + #[serde(rename = "ExternalDocumentInfo1", skip_serializing_if = "Option::is_none")] + external_document_info_1: Option, + + #[serde(rename = "ExternalDocumentInfo2", skip_serializing_if = "Option::is_none")] + external_document_info_2: Option, + + /// The file extension of the uploaded file. + #[serde(rename = "FileExtension", skip_serializing_if = "Option::is_none")] + file_extension: Option, + + /// The type of the file, such as PDF, PNG, WORD, or LINK. + #[serde(rename = "FileType", skip_serializing_if = "Option::is_none")] + file_type: Option, + + /// The ID of the first location where the file was published. + #[serde(rename = "FirstPublishLocationId", skip_serializing_if = "Option::is_none")] + first_publish_location_id: Option, + /* + /// Indicates whether the file is available externally. + #[serde(rename = "IsExternallyAvailable")] + is_externally_available: bool, + */ + /// Indicates whether this version is the latest version of the file. + #[serde(rename = "IsLatest", skip_serializing_if = "Option::is_none")] + is_latest: Option, + + /// Indicates whether this version is a major version. + #[serde(rename = "IsMajorVersion", skip_serializing_if = "Option::is_none")] + is_major_version: Option, + + /// The source from which the content originated. + #[serde(rename = "Origin", skip_serializing_if = "Option::is_none")] + origin: Option, + + /// The complete path of the file on the client machine when it was uploaded. + #[serde(rename = "PathOnClient", skip_serializing_if = "Option::is_none")] + path_on_client: Option, + + /// The reason this version of the file was created or changed. + #[serde(rename = "ReasonForChange", skip_serializing_if = "Option::is_none")] + reason_for_change: Option, + + /// Controls whether users can share the file with other users, groups, or records. + #[serde(rename = "SharingOption", skip_serializing_if = "Option::is_none")] + sharing_option: Option, + + /// Controls file privacy on records. + #[serde(rename = "SharingPrivacy", skip_serializing_if = "Option::is_none")] + sharing_privacy: Option, + + /// A comma-separated list of tags associated with the file. + #[serde(rename = "TagCsv", skip_serializing_if = "Option::is_none")] + tag_csv: Option, // todo - copy these over to new record + + /// A text preview of the file content. + #[serde(rename = "TextPreview", skip_serializing_if = "Option::is_none")] + text_preview: Option, + + /// The title of the file. + #[serde(rename = "Title", skip_serializing_if = "Option::is_none")] + title: Option, + + /// The body of the file, represented as base64-encoded content when retrieved through the API. + #[serde(rename = "VersionData", skip_serializing_if = "Option::is_none")] + version_data: Option, + + /// The version number of the file. + #[serde(rename = "VersionNumber", skip_serializing_if = "Option::is_none")] + version_number: Option, +} + +impl ContentVersion { + + pub fn content_document_id(&self) -> Option<&str> { + self.content_document_id.as_deref() + } + + pub fn first_publish_location_id(&self) -> Option<&str> { + self.first_publish_location_id.as_deref() + } + + pub fn query_fields() -> Vec<&'static str> { + vec![ + "Checksum", + "ContentDocumentId", + "ContentLocation", + "ContentSize", + "ContentUrl", + "Description", + "FileExtension", + "FileType", + "FirstPublishLocationId", + // "IsExternallyAvailable", + "Id", + "IsLatest", + "IsMajorVersion", + "Origin", + "OwnerId", + "PathOnClient", + "PublishStatus", + "ReasonForChange", + "SharingOption", + "SharingPrivacy", + "TagCsv", + "TextPreview", + "Title", + "VersionData", + "VersionNumber", + ] + } + + pub fn sobject_type() -> &'static str { + "ContentVersion" + } + + pub async fn query_binary_records(api: &mut RestApi) -> Result> { + let query = format!( + "SELECT {} FROM {} WHERE {} LIMIT 50", + Self::query_fields().join(","), + Self::sobject_type(), + "IsMajorVersion = TRUE AND ContentLocation = 'S' AND PublishStatus = 'P'", + ); + let result = Ok(api.query::(&query).await?.records); + result + } + + pub fn set_content_location(&mut self, value: &str) -> &mut Self { + self.content_location = Some(value.to_string()); + self + } + + pub fn set_content_url(&mut self, content_url: &str) -> &mut Self { + self.content_url = Some(content_url.to_string()); + self + } + + pub fn set_external_data_source_id(&mut self, value: &str) -> &mut Self { + self.external_data_source_id = Some(value.to_string()); + self + } + + pub fn set_external_document_info_1(&mut self, value: &str) -> &mut Self { + self.external_document_info_1 = Some(value.to_string()); + self + } + + pub fn set_external_document_info_2(&mut self, value: &str) -> &mut Self { + self.external_document_info_2 = Some(value.to_string()); + self + } + + pub fn set_first_publish_location_id(&mut self, value: &str) -> &mut Self { + self.first_publish_location_id = Some(value.to_string()); + self + } + pub fn set_sharing_option(&mut self, value: &str) -> &mut Self { + self.sharing_option = Some(value.to_string()); + self + } + pub fn set_sharing_privacy(&mut self, value: &str) -> &mut Self { + self.sharing_privacy = Some(value.to_string()); + self + } + + pub fn set_origin(&mut self, value: &str) -> &mut Self { + self.origin = Some(value.to_string()); + self + } + + pub fn set_is_major_version(&mut self, value: bool) -> &mut Self { + self.is_major_version = Some(value); + self + } + + pub fn set_title(&mut self, title: &str) -> &mut Self { + self.title = Some(title.to_string()); + self + } + + pub fn take_checksum(&mut self) -> Option { + self.checksum.take() + } + pub fn take_content_document_id(&mut self) -> Option { + self.content_document_id.take() + } + + pub fn take_content_size(&mut self) -> Option { + self.content_size.take() + } + + pub fn take_file_extension(&mut self) -> Option { + self.file_extension.take() + } + + pub fn take_file_type(&mut self) -> Option { + self.file_type.take() + } + + pub fn take_first_publish_location_id(&mut self) -> Option { + self.first_publish_location_id.take() + } + + pub fn take_id(&mut self) -> Option { + self.id.take() + } + + pub fn take_owner_id(&mut self) -> Option { + self.owner_id.take() + } + + pub fn take_origin(&mut self) -> Option { + self.origin.take() + } + + pub fn take_path_on_client(&mut self) -> Option { + self.path_on_client.take() + } + + pub fn take_sharing_option(&mut self) -> Option { + self.sharing_option.take() + } + + pub fn take_sharing_privacy(&mut self) -> Option { + self.sharing_privacy.take() + } + + pub fn take_title(&mut self) -> Option { + self.title.take() + } + + pub fn take_version_data(&mut self) -> Option { + self.version_data.take() + } + + pub fn take_version_number(&mut self) -> Option { + self.version_number.take() + } +} diff --git a/src/salesforce/models/external_data_source.rs b/src/salesforce/models/external_data_source.rs new file mode 100644 index 0000000..c779af9 --- /dev/null +++ b/src/salesforce/models/external_data_source.rs @@ -0,0 +1,5 @@ +use rustsf::DefSObject; + +#[DefSObject(sobject_type = "ExternalDataSource")] +pub(crate) struct ExternalDataSource { +} \ No newline at end of file diff --git a/src/salesforce/models/mod.rs b/src/salesforce/models/mod.rs new file mode 100644 index 0000000..3ce4c9d --- /dev/null +++ b/src/salesforce/models/mod.rs @@ -0,0 +1,7 @@ +mod content_version; +mod content_document_link; +mod external_data_source; + +pub(crate) use content_version::ContentVersion; +pub(crate) use content_document_link::ContentDocumentLink; +pub(crate) use external_data_source::ExternalDataSource; diff --git a/src/terminator.rs b/src/terminator.rs index fb2dd68..af1cc8e 100644 --- a/src/terminator.rs +++ b/src/terminator.rs @@ -14,7 +14,7 @@ pub struct Terminator { impl Terminator { pub(crate) fn is_fired(&self) -> bool { - !self.term.load(Ordering::Relaxed) + self.term.load(Ordering::Relaxed) } }