WIP

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",

Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "imojefs"
+name = "imoje"
 description = "Johnson & Johnson IMoje CLI"
 authors = ["Wim Velzeboer <wim@velzeboer.nl>"]
 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"

src/aws_s3.rs

@@ -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)
-}

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<String>,
 
 		/// 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<String>,
 
 		/// 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<String>,
 
 		/// 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<String>,
 
 		/// 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<String>,
 
 		/// 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<String>,
 
 		/// 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<String>,
+
+		/// 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) {
+	let app_config =
-		Ok(()) => {}
+		crate::config::load(&cli.config)
-		Err(e) => {
+			.context("Failed to load application configuration")?
-			error!("Error loading configuration file: {:#}", e);
+			.new_file_sync()
-			std::process::exit(1);
+			.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<String>,
 	s3_region: &Option<String>,
 	sfdx_auth_url: &Option<String>,
+	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");

src/cli/set/mod.rs

@@ -7,7 +7,9 @@ pub(crate) fn run(config_file: &str,
                   s3_client_secret: &Option<String>,
                   s3_endpoint: &Option<String>,
                   s3_region: &Option<String>,
-                  sfdx_auth_url: &Option<String>,) -> Result<()> {
+                  sfdx_auth_url: &Option<String>,
+                  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")?;
 

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 anyhow::{Context, Result, anyhow};
-use base64::engine::general_purpose::STANDARD as BASE64;
 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 = "<not set>";
@@ -25,6 +25,7 @@ pub struct AppConfig {
 	pub s3_endpoint: Option<String>,
 	pub s3_region: Option<String>,
 	pub sfdx_auth_url: Option<String>,
+	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<crate::file_sync::FileSyncConfig> {
+		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!(
+	println!("admin_email = {}", app_config.admin_email.as_deref().unwrap_or(NOT_SET));
-		"admin_email = {}",
+	println!("s3_bucket = {}", app_config.s3_bucket.as_deref().unwrap_or(NOT_SET));
-		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!(
+	println!("s3_endpoint = {}", app_config.s3_endpoint.as_deref().unwrap_or(NOT_SET));
-		"s3_endpoint = {}",
-		app_config.s3_endpoint.as_deref().unwrap_or(NOT_SET)
-	);
 
-	println!(
+	println!("s3_region = {}", app_config.s3_region.as_deref().unwrap_or(NOT_SET));
-		"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<AppConfig> {
@@ -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)
+	let serialized_config = toml::to_string_pretty(config).context("Unable to serialize default configuration")?;
-		.context("Unable to serialize default configuration")?;
 
-	let encrypted_config = encrypt_config(&serialized_config)
+	let encrypted_config = encrypt_config(&serialized_config).context("Unable to encrypt default configuration")?;
-		.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<String> {
 	let key = config_key()?;
-	let cipher = Aes256Gcm::new_from_slice(&key)
+	let cipher = Aes256Gcm::new_from_slice(&key).map_err(|_| anyhow!("Invalid encryption key length"))?;
-		.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<AppConfig> {
 	}
 
 	let key = config_key()?;
-	let cipher = Aes256Gcm::new_from_slice(&key)
+	let cipher = Aes256Gcm::new_from_slice(&key).map_err(|_| anyhow!("Invalid encryption key length"))?;
-		.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<AppConfig> {
 		.decrypt(nonce, ciphertext)
 		.map_err(|error| anyhow!("Unable to decrypt configuration: {error}"))?;
 
-	let plaintext = String::from_utf8(plaintext)
+	let plaintext = String::from_utf8(plaintext).context("Configuration file does not contain valid UTF-8")?;
-		.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)
+	let encoded_key =
-		.unwrap_or("ihDlaCqJqUGJ1am4dT9k/BHRG2UT+iK9IfYcPe0Db/I=".to_string());
+		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'")
+		let encrypted = encrypt_config("sfdx_auth_url = 'auth-url'").expect("configuration should encrypt");
-			.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();
 	}
-}
+}

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<String>,
+	pub s3_bucket: String,
+	pub s3_client_id: String,
+	pub s3_client_secret: String,
+	pub s3_endpoint: Option<String>,
+	pub s3_region: Option<String>,
+	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
+	}
+}
+

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<ContentDocumentLink>,
+}
+impl Document {
+
+	pub(crate) fn content_version(&self) -> &ContentVersion {
+		&self.content_version
+	}
+
+	pub(crate) fn from_content_versions_and_links(
+		content_versions: Vec<ContentVersion>,
+		content_document_links: Vec<ContentDocumentLink>,
+	) -> Vec<Document> {
+		let mut links_by_content_document_id: HashMap<String, Vec<ContentDocumentLink>> =
+			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<ContentDocumentLink>) {
+		(self.content_version, self.content_document_links)
+	}
+
+}
+
+pub async fn new_batch(mut sf_client: &mut RestApi) -> Result<Vec<Document>> {
+	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)
+}

src/file_sync/mod.rs

@@ -1,37 +1,238 @@
-pub fn run() {
+use anyhow::{Context, Result, anyhow};
-	unimplemented!("File sync not yet implemented");
+use bytes::Bytes;
+use log::{error, info};
+use mime_type::{MimeFormat, MimeType};
+use pluralizer::pluralize;
+use rustsf::{RestApi, SObject};
+use sha2::{Digest, Sha256};
 
-	/*
+pub(crate) mod config;
-let args = cli::Args::parse();
 
+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;
 
-
+async fn sleep(wait: u64) {
-let salesforce_client = salesforce::connect(&args.sfdx_auth_url).await?;
+	info!("Waiting {} seconds...", wait);
-let s3_client = aws_s3::connect(
+	tokio::time::sleep(std::time::Duration::from_secs(wait)).await; // every 30 seconds
-	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...");
 }
 
+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.");*/
+		sleep(config.wait()).await;
-	
+	}
-}
+
+	info!("Done...");
+
+	Ok(())
+}
+
+async fn run_batch(config: &config::Config) -> Result<usize> {
+	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<String> {
+	let query = "SELECT Id FROM ExternalDataSource WHERE DeveloperName = 'S3_EmailAttachmentStorage' WITH SYSTEM_MODE LIMIT 1";
+	let eds_id = sf_client.query::<ExternalDataSource>(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<ContentVersion> {
+	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::<ContentVersion>(&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<Bytes> {
+	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))
+}

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<Self> {
+		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"));
+	}
+}

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<config::AppConfig> = OnceLock::new();
 
-#[::tokio::main]
+fn main() {
-async fn main() {
+
     cli::run();
 }

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<aws_sdk_s3::operation::put_object::PutObjectOutput> {
+	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:
+///   `"<bucket-name>/<object-key>"`
+/// * 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<aws_sdk_s3::operation::get_object::GetObjectOutput, S3Error> {
+	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<aws_sdk_s3::operation::put_object::PutObjectOutput, S3Error> {
+	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<dyn std::error::Error>> {
+///     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<String>` 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<Vec<String>, 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<String> = 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<dyn Error>> {
+///     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<String>,
+) -> Result<(), S3Error> {
+	// Push into a mut vector to use `?` early return errors while building object keys.
+	let mut delete_object_ids: Vec<aws_sdk_s3::types::ObjectIdentifier> = 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", &region).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<Option<aws_sdk_s3::operation::create_bucket::CreateBucketOutput>, 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#"<Error><Code>NoSuchBucket</Code><Message>The specified bucket does not exist</Message><BucketName>bucket_name</BucketName><RequestId>REQUEST</RequestId><HostId>HOSTID=</HostId></Error>"#
+		);
+
+		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#"<?xml version="1.0" encoding="UTF-8"?><ListBucketResult>
+                            <Name>test</Name>
+                            <Contents><Key>obj1</Key></Contents>
+                            <Contents><Key>obj2</Key></Contents>
+                            <KeyCount>2</KeyCount>
+                            </ListBucketResult>"#
+						)
+					),
+					// client.delete_objects().delete(...(delete_objects)...))
+					test_event!(r#""#, (200, r#"<?xml version="1.0" encoding="UTF-8"?>
+                    <DeleteResult>
+                        <Deleted>
+                            <DeleteMarker>true</DeleteMarker>
+                            <Key>obj1</Key>
+                        </Deleted>
+                        <Deleted>
+                            <DeleteMarker>true</DeleteMarker>
+                            <Key>obj2</Key>
+                        </Deleted>
+                    </DeleteResult>
+                    "#)),
+					// client.list_objects_v2().bucket(bucket_name)
+					test_event!(
+						r#""#,
+						(
+							200,
+							r#"<?xml version="1.0" encoding="UTF-8"?><ListBucketResult><Name>test</Name>
+                            <KeyCount>0</KeyCount>
+                            </ListBucketResult>"#
+						)
+					),
+				]))
+				.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#"<?xml version="1.0" encoding="UTF-8"?><ListBucketResult>
+                            <Name>test</Name>
+                            <Contents><Key>obj1</Key></Contents>
+                            <Contents><Key>obj2</Key></Contents>
+                            <KeyCount>2</KeyCount>
+                            </ListBucketResult>"#
+						)
+					),
+					// client.delete_objects().delete(...(delete_objects)...))
+					test_event!(
+						r#""#,
+						(
+							200,
+							r#"<?xml version="1.0" encoding="UTF-8"?>
+                    <DeleteResult>
+                        <Deleted>
+                            <DeleteMarker>true</DeleteMarker>
+                            <Key>obj1</Key>
+                        </Deleted>
+                        <Deleted>
+                            <DeleteMarker>true</DeleteMarker>
+                            <Key>obj2</Key>
+                        </Deleted>
+                    </DeleteResult>
+                    "#
+						)
+					),
+					// client.list_objects_v2().bucket(bucket_name)
+					test_event!(
+						r#""#,
+						(
+							200,
+							r#"<?xml version="1.0" encoding="UTF-8"?><ListBucketResult>
+                            <Name>test</Name>
+                            <Contents><Key>obj3</Key></Contents>
+                            <KeyCount>1</KeyCount>
+                            </ListBucketResult>"#
+						)
+					),
+				]))
+				.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#"<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult>
+   <Name>test</Name>
+</ListBucketResult>"#
+		);
+
+		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"));
+	}
+}
+*/

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)
+}

src/salesforce/mod.rs

@@ -12,4 +12,24 @@ pub async fn connect(sfdx_auth_url: &str) -> Result<RestApi> {
 		.context("Unable to connect to Salesforce")?;
 
 	Ok(RestApi::new(client))
-}
+}
+
+/*
+pub(crate) fn retrieve_binary_content_docs(api: RestApi) -> Vec<models::ContentVersion> {
+/*
+	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()*/
+
+}*/

src/salesforce/models.rs

@@ -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<String>,
-
-    /// The location where the content is stored. Valid values include Salesforce, external, or linked content locations.
-    #[serde(rename = "ContentLocation")]
-    content_location: Option<String>,
-
-    /// The date and time when the file content was last modified.
-    #[serde(rename = "ContentModifiedDate")]
-    content_modified_date: Option<String>,
-
-    /// The ID of the user who last modified the file content.
-    #[serde(rename = "ContentModifiedById")]
-    content_modified_by_id: Option<String>,
-
-    /// The URL of the content when the file is stored outside Salesforce.
-    #[serde(rename = "ContentUrl")]
-    content_url: Option<String>,
-
-    /// The version number of the file.
-    #[serde(rename = "VersionNumber")]
-    version_number: Option<String>,
-
-    /// The title of the file.
-    #[serde(rename = "Title")]
-    title: Option<String>,
-
-    /// The description of the file.
-    #[serde(rename = "Description")]
-    description: Option<String>,
-
-    /// The reason this version of the file was created or changed.
-    #[serde(rename = "ReasonForChange")]
-    reason_for_change: Option<String>,
-
-    /// Controls whether users can share the file with other users, groups, or records.
-    #[serde(rename = "SharingOption")]
-    sharing_option: Option<String>,
-
-    /// Controls file privacy on records.
-    #[serde(rename = "SharingPrivacy")]
-    sharing_privacy: Option<String>,
-
-    /// The complete path of the file on the client machine when it was uploaded.
-    #[serde(rename = "PathOnClient")]
-    path_on_client: Option<String>,
-
-    /// 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<String>,
-
-    /// External document information used to identify externally stored content.
-    #[serde(rename = "ExternalDocumentInfo1")]
-    external_document_info1: Option<String>,
-
-    /// Additional external document information used to identify externally stored content.
-    #[serde(rename = "ExternalDocumentInfo2")]
-    external_document_info2: Option<String>,
-
-    /// The boost value used to promote featured content.
-    #[serde(rename = "FeaturedContentBoost")]
-    featured_content_boost: Option<i64>,
-
-    /// The date when the file was marked as featured content.
-    #[serde(rename = "FeaturedContentDate")]
-    featured_content_date: Option<String>,
-
-    /// The file extension of the uploaded file.
-    #[serde(rename = "FileExtension")]
-    file_extension: Option<String>,
-
-    /// The type of the file, such as PDF, PNG, WORD, or LINK.
-    #[serde(rename = "FileType")]
-    file_type: Option<String>,
-
-    /// The ID of the first location where the file was published.
-    #[serde(rename = "FirstPublishLocationId")]
-    first_publish_location_id: Option<String>,
-
-    /// The source from which the content originated.
-    #[serde(rename = "Origin")]
-    origin: Option<String>,
-
-    /// The ID of the Experience Cloud site or network associated with the file.
-    #[serde(rename = "NetworkId")]
-    network_id: Option<String>,
-
-    /// The number of positive ratings submitted for this file version.
-    #[serde(rename = "PositiveRatingCount")]
-    positive_rating_count: Option<i64>,
-
-    /// The publishing status of the file version.
-    #[serde(rename = "PublishStatus")]
-    publish_status: Option<String>,
-
-    /// A comma-separated list of tags associated with the file.
-    #[serde(rename = "TagCsv")]
-    tag_csv: Option<String>,
-
-    /// A text preview of the file content.
-    #[serde(rename = "TextPreview")]
-    text_preview: Option<String>,
-
-    /// The body of the file, represented as base64-encoded content when retrieved through the API.
-    #[serde(rename = "VersionData")]
-    version_data: Option<String>,
-
-    /// The size of the file in bytes.
-    #[serde(rename = "ContentSize")]
-    content_size: Option<i64>,
-
-    /// MD5 checksum for the file.
-    #[serde(rename = "Checksum")]
-    checksum: Option<String>,
-
-    /// The ID of the ContentBody record that stores the binary file data.
-    #[serde(rename = "ContentBodyId")]
-    content_body_id: Option<String>,
-
-    /// 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,
-
-}

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<Vec<ContentDocumentLink>> {
+		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::<Vec<_>>()
+					.join(",")
+			)
+		);
+		let result = Ok(api.query::<ContentDocumentLink>(&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
+	}
+}

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<String>,
+
+	/// The ID of the ContentDocument associated with this version.
+	#[serde(rename = "ContentDocumentId", skip_serializing_if = "Option::is_none")]
+	content_document_id: Option<String>,
+
+	/// 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<String>,
+
+	/// The size of the file in bytes.
+	#[serde(rename = "ContentUrl", skip_serializing_if = "Option::is_none")]
+	content_url: Option<String>,
+
+	/// The size of the file in bytes.
+	#[serde(rename = "ContentSize", skip_serializing_if = "Option::is_none")]
+	content_size: Option<u64>,
+
+	/// The description of the file.
+	#[serde(rename = "Description", skip_serializing_if = "Option::is_none")]
+	description: Option<String>,
+
+	#[serde(rename = "ExternalDataSourceId", skip_serializing_if = "Option::is_none")]
+	external_data_source_id: Option<String>,
+
+	#[serde(rename = "ExternalDocumentInfo1", skip_serializing_if = "Option::is_none")]
+	external_document_info_1: Option<String>,
+
+	#[serde(rename = "ExternalDocumentInfo2", skip_serializing_if = "Option::is_none")]
+	external_document_info_2: Option<String>,
+
+	/// The file extension of the uploaded file.
+	#[serde(rename = "FileExtension", skip_serializing_if = "Option::is_none")]
+	file_extension: Option<String>,
+
+	/// The type of the file, such as PDF, PNG, WORD, or LINK.
+	#[serde(rename = "FileType", skip_serializing_if = "Option::is_none")]
+	file_type: Option<String>,
+
+	/// 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<String>,
+	/*
+		/// 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<bool>,
+
+	/// Indicates whether this version is a major version.
+	#[serde(rename = "IsMajorVersion", skip_serializing_if = "Option::is_none")]
+	is_major_version: Option<bool>,
+
+	/// The source from which the content originated.
+	#[serde(rename = "Origin", skip_serializing_if = "Option::is_none")]
+	origin: Option<String>,
+
+	/// 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<String>,
+
+	/// The reason this version of the file was created or changed.
+	#[serde(rename = "ReasonForChange", skip_serializing_if = "Option::is_none")]
+	reason_for_change: Option<String>,
+
+	/// 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<String>,
+
+	/// Controls file privacy on records.
+	#[serde(rename = "SharingPrivacy", skip_serializing_if = "Option::is_none")]
+	sharing_privacy: Option<String>,
+
+	/// A comma-separated list of tags associated with the file.
+	#[serde(rename = "TagCsv", skip_serializing_if = "Option::is_none")]
+	tag_csv: Option<String>, // 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<String>,
+
+	/// The title of the file.
+	#[serde(rename = "Title", skip_serializing_if = "Option::is_none")]
+	title: Option<String>,
+
+	/// 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<String>,
+
+	/// The version number of the file.
+	#[serde(rename = "VersionNumber", skip_serializing_if = "Option::is_none")]
+	version_number: Option<String>,
+}
+
+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<Vec<ContentVersion>> {
+		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::<ContentVersion>(&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<String> {
+		self.checksum.take()
+	}
+	pub fn take_content_document_id(&mut self) -> Option<String> {
+		self.content_document_id.take()
+	}
+
+	pub fn take_content_size(&mut self) -> Option<u64> {
+		self.content_size.take()
+	}
+
+	pub fn take_file_extension(&mut self) -> Option<String> {
+		self.file_extension.take()
+	}
+
+	pub fn take_file_type(&mut self) -> Option<String> {
+		self.file_type.take()
+	}
+
+	pub fn take_first_publish_location_id(&mut self) -> Option<String> {
+		self.first_publish_location_id.take()
+	}
+
+	pub fn take_id(&mut self) -> Option<String> {
+		self.id.take()
+	}
+
+	pub fn take_owner_id(&mut self) -> Option<String> {
+		self.owner_id.take()
+	}
+
+	pub fn take_origin(&mut self) -> Option<String> {
+		self.origin.take()
+	}
+
+	pub fn take_path_on_client(&mut self) -> Option<String> {
+		self.path_on_client.take()
+	}
+
+	pub fn take_sharing_option(&mut self) -> Option<String> {
+		self.sharing_option.take()
+	}
+
+	pub fn take_sharing_privacy(&mut self) -> Option<String> {
+		self.sharing_privacy.take()
+	}
+
+	pub fn take_title(&mut self) -> Option<String> {
+		self.title.take()
+	}
+
+	pub fn take_version_data(&mut self) -> Option<String> {
+		self.version_data.take()
+	}
+
+	pub fn take_version_number(&mut self) -> Option<String> {
+		self.version_number.take()
+	}
+}

src/salesforce/models/external_data_source.rs

@@ -0,0 +1,5 @@
+use rustsf::DefSObject;
+
+#[DefSObject(sobject_type = "ExternalDataSource")]
+pub(crate) struct ExternalDataSource {
+}

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;

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)
     }
 }