Skip to main content

rustfs_cli/commands/
cp.rs

1//! cp command - Copy objects
2//!
3//! Copies objects between local filesystem and S3, or between S3 locations.
4
5use clap::Args;
6use jiff::Timestamp;
7use rc_core::alias::RetryConfig;
8use rc_core::{
9    AliasManager, Error, MetadataDirective, MultipartCopyCancellation, MultipartCopyOptions,
10    ObjectAttributes, ObjectEncryptionRequest, ObjectInfo, ObjectKeyPolicy, ObjectStore as _,
11    ObjectWriteOptions, ParsedPath, RemotePath, SseCustomerKey, TransferCancellation,
12    TransferCandidate, TransferControls, TransferCopyOptions, TransferExecutor,
13    TransferOutcomeState, TransferPlan, TransferReadOptions, TransferSelection,
14    normalize_relative_key, parse_path, relative_local_path_from_key,
15};
16use rc_s3::S3Client;
17use serde::Serialize;
18use std::collections::{BTreeSet, HashMap, HashSet};
19use std::fmt;
20use std::path::{Path, PathBuf};
21use std::sync::{Arc, Mutex as StdMutex};
22use tokio::sync::Mutex as AsyncMutex;
23
24use crate::exit_code::ExitCode;
25use crate::output::{Formatter, OutputConfig, ProgressBar, V3SuccessEnvelope};
26use crate::secret_input::{SecretLocator, resolve_secret_locator};
27
28use super::object_identity::set_source_identity;
29use super::transfer_fidelity::{MetadataDirectiveArg, TaggingDirectiveArg, TransferFidelityArgs};
30
31const CP_AFTER_HELP: &str = "\
32Examples:
33  rc object copy ./report.json local/my-bucket/reports/
34  rc cp ./report.json local/my-bucket/reports/
35  rc object copy local/source-bucket/archive.tar.gz ./downloads/archive.tar.gz";
36
37pub(crate) const GET_AFTER_HELP: &str = "\
38Examples:
39  rc get local/my-bucket/report.json ./report.json
40  rc get local/my-bucket/archive.tar.gz ./downloads/archive.tar.gz";
41
42pub(crate) const PUT_AFTER_HELP: &str = "\
43Examples:
44  rc put ./report.json local/my-bucket/reports/
45  rc put ./january.csv ./february.csv local/my-bucket/reports/";
46
47const REMOTE_PATH_SUGGESTION: &str =
48    "Use a local filesystem path or a remote path in the form alias/bucket[/key].";
49const DEFAULT_TRANSFER_CONCURRENCY: usize = 4;
50const DEFAULT_RETRY_ATTEMPTS: u32 = 3;
51const DEFAULT_RETRY_INITIAL_BACKOFF_MS: u64 = 100;
52const DEFAULT_RETRY_MAX_BACKOFF_MS: u64 = 10_000;
53#[cfg(test)]
54const MAX_SINGLE_COPY_SIZE: u64 = rc_core::S3_SINGLE_COPY_MAX_SIZE;
55
56/// Copy objects
57#[derive(Args, Clone)]
58#[command(after_help = CP_AFTER_HELP)]
59pub struct CpArgs {
60    /// Source paths (local paths or alias/bucket/key)
61    #[arg(required = true, num_args = 1.., value_name = "SOURCE")]
62    pub sources: Vec<String>,
63
64    /// Destination path (local path or alias/bucket/key)
65    pub target: String,
66
67    /// Copy recursively
68    #[arg(short, long)]
69    pub recursive: bool,
70
71    /// Preserve file attributes
72    #[arg(short, long, conflicts_with = "metadata_directive")]
73    pub preserve: bool,
74
75    /// Source metadata handling for remote copies
76    #[arg(long, value_enum)]
77    pub(crate) metadata_directive: Option<MetadataDirectiveArg>,
78
79    /// Source tag handling for remote copies
80    #[arg(long, value_enum)]
81    pub(crate) tagging_directive: Option<TaggingDirectiveArg>,
82
83    /// Continue on errors
84    #[arg(long)]
85    pub continue_on_error: bool,
86
87    /// Overwrite destination if it exists
88    #[arg(
89        long,
90        default_value_t = true,
91        action = clap::ArgAction::Set,
92        num_args = 0..=1,
93        default_missing_value = "true"
94    )]
95    pub overwrite: bool,
96
97    /// Skip remote destinations that already exist
98    #[arg(long)]
99    pub skip_existing: bool,
100
101    /// Only show what would be copied (dry run)
102    #[arg(long)]
103    pub dry_run: bool,
104
105    /// Storage class for destination (S3 only)
106    #[arg(long)]
107    pub storage_class: Option<String>,
108
109    /// Content type for uploaded files
110    #[arg(long)]
111    pub content_type: Option<String>,
112
113    #[command(flatten)]
114    pub(crate) fidelity: TransferFidelityArgs,
115
116    /// Apply SSE-S3 to the remote destination path
117    #[arg(long = "enc-s3")]
118    pub enc_s3: Vec<String>,
119
120    /// Apply SSE-KMS to the remote destination path as TARGET=KMS_KEY_ID
121    #[arg(long = "enc-kms")]
122    pub enc_kms: Vec<String>,
123
124    /// Read a 32-byte SSE-C source key from a protected file
125    #[arg(long = "enc-c-source-key-file")]
126    pub enc_c_source_key_file: Option<PathBuf>,
127
128    /// Read a 32-byte SSE-C source key from the named environment variable
129    #[arg(long = "enc-c-source-key-env")]
130    pub enc_c_source_key_env: Option<String>,
131
132    /// Read a 32-byte SSE-C destination key from a protected file
133    #[arg(long = "enc-c-destination-key-file")]
134    pub enc_c_destination_key_file: Option<PathBuf>,
135
136    /// Read a 32-byte SSE-C destination key from the named environment variable
137    #[arg(long = "enc-c-destination-key-env")]
138    pub enc_c_destination_key_env: Option<String>,
139
140    #[arg(skip)]
141    pub(crate) source_customer_key: Option<SseCustomerKey>,
142
143    #[arg(skip)]
144    pub(crate) destination_customer_key: Option<SseCustomerKey>,
145
146    /// Include source-relative paths matching this glob (repeatable)
147    #[arg(long)]
148    pub include: Vec<String>,
149
150    /// Exclude source-relative paths matching this glob; exclusions always win (repeatable)
151    #[arg(long)]
152    pub exclude: Vec<String>,
153
154    /// Select objects modified more recently than this age (for example 1h or 7d)
155    #[arg(long)]
156    pub newer_than: Option<String>,
157
158    /// Select objects modified less recently than this age (for example 1h or 7d)
159    #[arg(long)]
160    pub older_than: Option<String>,
161
162    /// Select object state at or before a UTC timestamp or age
163    #[arg(long)]
164    pub rewind: Option<String>,
165
166    /// Maximum number of transfers in flight across the command
167    #[arg(long)]
168    pub concurrency: Option<usize>,
169
170    /// Aggregate transfer start rate in bytes per second (for example 10MiB/s)
171    #[arg(long)]
172    pub rate_limit: Option<String>,
173
174    /// Maximum attempts for transient transfer failures
175    #[arg(long)]
176    pub retry_attempts: Option<u32>,
177
178    /// Initial transient retry backoff in milliseconds
179    #[arg(long)]
180    pub retry_initial_backoff_ms: Option<u64>,
181
182    /// Maximum transient retry backoff in milliseconds
183    #[arg(long)]
184    pub retry_max_backoff_ms: Option<u64>,
185
186    /// Return not-found when selection produces no transfer candidates
187    #[arg(long)]
188    pub fail_empty: bool,
189
190    /// Print deterministic aggregate transfer counters (human output)
191    #[arg(long)]
192    pub summary: bool,
193
194    /// Reject object keys that cannot be created on Windows filesystems
195    #[arg(long)]
196    pub portable_names: bool,
197}
198
199impl fmt::Debug for CpArgs {
200    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
201        formatter.write_str("CpArgs { .. }")
202    }
203}
204
205impl CpArgs {
206    pub(crate) fn single(source: impl Into<String>, target: impl Into<String>) -> Self {
207        Self {
208            sources: vec![source.into()],
209            target: target.into(),
210            recursive: false,
211            preserve: false,
212            metadata_directive: None,
213            tagging_directive: None,
214            continue_on_error: false,
215            overwrite: true,
216            skip_existing: false,
217            dry_run: false,
218            storage_class: None,
219            content_type: None,
220            fidelity: TransferFidelityArgs::default(),
221            enc_s3: Vec::new(),
222            enc_kms: Vec::new(),
223            enc_c_source_key_file: None,
224            enc_c_source_key_env: None,
225            enc_c_destination_key_file: None,
226            enc_c_destination_key_env: None,
227            source_customer_key: None,
228            destination_customer_key: None,
229            include: Vec::new(),
230            exclude: Vec::new(),
231            newer_than: None,
232            older_than: None,
233            rewind: None,
234            concurrency: None,
235            rate_limit: None,
236            retry_attempts: None,
237            retry_initial_backoff_ms: None,
238            retry_max_backoff_ms: None,
239            fail_empty: false,
240            summary: false,
241            portable_names: false,
242        }
243    }
244
245    fn local_key_policy(&self) -> ObjectKeyPolicy {
246        ObjectKeyPolicy::for_local_destination(self.portable_names)
247    }
248}
249
250/// Download one remote object through the canonical copy implementation.
251#[derive(Args, Debug)]
252#[command(
253    override_usage = "rc get [OPTIONS] <SOURCE> <TARGET>",
254    after_help = GET_AFTER_HELP
255)]
256pub struct GetArgs {
257    #[command(flatten)]
258    pub transfer: CpArgs,
259}
260
261/// Upload one or more local paths through the canonical copy implementation.
262#[derive(Args, Debug)]
263#[command(after_help = PUT_AFTER_HELP)]
264pub struct PutArgs {
265    #[command(flatten)]
266    pub transfer: CpArgs,
267}
268
269#[derive(Clone, Copy, Debug, Eq, PartialEq)]
270enum TransferAlias {
271    Copy,
272    Get,
273    Put,
274}
275
276#[derive(Debug, Serialize)]
277struct CpOutput {
278    status: &'static str,
279    source: String,
280    target: String,
281    #[serde(skip_serializing_if = "Option::is_none")]
282    size_bytes: Option<i64>,
283    #[serde(skip_serializing_if = "Option::is_none")]
284    size_human: Option<String>,
285    #[serde(skip_serializing_if = "Option::is_none")]
286    version_id: Option<String>,
287    #[serde(skip_serializing_if = "Option::is_none")]
288    source_version_id: Option<String>,
289}
290
291#[derive(Debug, Serialize)]
292struct VersionCopyData {
293    operation: &'static str,
294    source: String,
295    target: String,
296    source_version_id: Option<String>,
297    version_id: Option<String>,
298    size_bytes: Option<i64>,
299    size_human: Option<String>,
300}
301
302/// Execute the cp command
303pub async fn execute(args: CpArgs, output_config: OutputConfig) -> ExitCode {
304    execute_with_alias(args, output_config, TransferAlias::Copy).await
305}
306
307/// Execute the mc-compatible `get` command through the canonical copy path.
308pub async fn execute_get(args: GetArgs, output_config: OutputConfig) -> ExitCode {
309    execute_with_alias(args.transfer, output_config, TransferAlias::Get).await
310}
311
312/// Execute the mc-compatible `put` command through the canonical copy path.
313pub async fn execute_put(args: PutArgs, output_config: OutputConfig) -> ExitCode {
314    execute_with_alias(args.transfer, output_config, TransferAlias::Put).await
315}
316
317async fn execute_with_alias(
318    mut args: CpArgs,
319    output_config: OutputConfig,
320    alias: TransferAlias,
321) -> ExitCode {
322    let formatter = Formatter::new(output_config);
323
324    if alias == TransferAlias::Get && args.sources.len() != 1 {
325        return formatter.fail(
326            ExitCode::UsageError,
327            "get requires exactly one remote source and one local target",
328        );
329    }
330
331    if let Err(error) = validate_destination_storage_class(args.storage_class.as_deref()) {
332        return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
333    }
334    if let Err(error) = object_write_options(
335        &args.fidelity,
336        args.content_type.as_deref(),
337        None,
338        None,
339        args.storage_class.clone(),
340    ) {
341        return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
342    }
343
344    if formatter.is_json() && uses_transfer_planner(&args) {
345        return formatter.fail(
346            ExitCode::UnsupportedFeature,
347            "Bulk transfer planning currently supports human output only; JSON batch output requires the versioned output contract",
348        );
349    }
350
351    let selection = match build_transfer_selection(&args, Timestamp::now()) {
352        Ok(selection) => selection,
353        Err(error) => return formatter.fail(ExitCode::UsageError, &error),
354    };
355    let controls = match build_transfer_controls(&args) {
356        Ok(controls) => controls,
357        Err(error) => return formatter.fail(ExitCode::UsageError, &error),
358    };
359
360    let alias_manager = AliasManager::new();
361
362    // Parse every operand before starting any transfer.
363    let mut sources = Vec::with_capacity(args.sources.len());
364    for source in &args.sources {
365        let parsed = match alias {
366            // `put` defines every source operand as a local path, including
367            // relative paths containing a slash that do not exist yet.
368            TransferAlias::Put => Ok(ParsedPath::Local(PathBuf::from(source))),
369            TransferAlias::Copy | TransferAlias::Get => {
370                parse_cp_path(source, alias_manager.as_ref().ok())
371            }
372        };
373        match parsed {
374            Ok(path) => sources.push(path),
375            Err(error) => {
376                return formatter.fail_with_suggestion(
377                    ExitCode::UsageError,
378                    &format!("Invalid source path '{source}': {error}"),
379                    REMOTE_PATH_SUGGESTION,
380                );
381            }
382        }
383    }
384
385    let parsed_target = match alias {
386        // `get` defines its target operand as a local path. Do not reinterpret
387        // a non-existent `directory/file` path as an alias and bucket.
388        TransferAlias::Get => Ok(ParsedPath::Local(PathBuf::from(&args.target))),
389        TransferAlias::Copy | TransferAlias::Put => {
390            parse_cp_path(&args.target, alias_manager.as_ref().ok())
391        }
392    };
393    let target = match parsed_target {
394        Ok(p) => p,
395        Err(e) => {
396            return formatter.fail_with_suggestion(
397                ExitCode::UsageError,
398                &format!("Invalid target path: {e}"),
399                REMOTE_PATH_SUGGESTION,
400            );
401        }
402    };
403    if let Err(error) = validate_alias_direction(alias, &sources, &target) {
404        return formatter.fail(ExitCode::UsageError, error);
405    }
406    if let Err(error) = validate_fidelity_directions(&args, &sources, &target) {
407        return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
408    }
409    let source_key_locator = match resolve_secret_locator(
410        args.enc_c_source_key_file.clone(),
411        args.enc_c_source_key_env.clone(),
412    ) {
413        Ok(locator) => locator,
414        Err(error) => {
415            return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
416        }
417    };
418    let destination_key_locator = match resolve_secret_locator(
419        args.enc_c_destination_key_file.clone(),
420        args.enc_c_destination_key_env.clone(),
421    ) {
422        Ok(locator) => locator,
423        Err(error) => {
424            return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
425        }
426    };
427    if source_key_locator.is_some()
428        && sources
429            .iter()
430            .any(|path| matches!(path, ParsedPath::Local(_)))
431    {
432        return formatter.fail(
433            ExitCode::UsageError,
434            "SSE-C source keys require every source to be remote",
435        );
436    }
437    if destination_key_locator.is_some() && matches!(target, ParsedPath::Local(_)) {
438        return formatter.fail(
439            ExitCode::UsageError,
440            "SSE-C destination keys require a remote destination",
441        );
442    }
443    if args.storage_class.is_some() && matches!(target, ParsedPath::Local(_)) {
444        return formatter.fail(
445            ExitCode::UsageError,
446            "--storage-class requires a remote destination",
447        );
448    }
449
450    let target_is_container = is_container_target(&args.target, &target);
451    if args.sources.len() > 1 && !target_is_container {
452        return formatter.fail(
453            ExitCode::UsageError,
454            "Multiple copy sources require a directory or remote prefix destination ending in '/'",
455        );
456    }
457
458    let target_encryption = match parse_destination_encryption(&args.enc_s3, &args.enc_kms, &target)
459    {
460        Ok(encryption) => encryption,
461        Err(error) => return formatter.fail(ExitCode::UsageError, &error),
462    };
463    if destination_key_locator.is_some() && target_encryption.is_some() {
464        return formatter.fail(
465            ExitCode::UsageError,
466            "SSE-C destination keys cannot be combined with --enc-s3 or --enc-kms",
467        );
468    }
469    if !args.dry_run
470        && (source_key_locator.is_some() || destination_key_locator.is_some())
471        && sources
472            .iter()
473            .all(|path| matches!(path, ParsedPath::Remote(_)))
474        && matches!(target, ParsedPath::Remote(_))
475    {
476        return formatter.fail(
477            ExitCode::UnsupportedFeature,
478            "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467",
479        );
480    }
481    if !args.dry_run {
482        args.source_customer_key = match load_customer_key(source_key_locator.as_ref()) {
483            Ok(key) => key,
484            Err(error) => {
485                return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
486            }
487        };
488        args.destination_customer_key = match load_customer_key(destination_key_locator.as_ref()) {
489            Ok(key) => key,
490            Err(error) => {
491                return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
492            }
493        };
494    }
495
496    let single_local_to_local = matches!(
497        (sources.as_slice(), &target),
498        ([ParsedPath::Local(_)], ParsedPath::Local(_))
499    );
500    if uses_transfer_planner(&args) && !single_local_to_local {
501        let alias_manager = match alias_manager.as_ref() {
502            Ok(manager) => manager,
503            Err(error) => {
504                return formatter.fail(
505                    ExitCode::UsageError,
506                    &format!("Failed to load aliases: {error}"),
507                );
508            }
509        };
510        return execute_transfer_plan(
511            &args,
512            sources,
513            target,
514            target_is_container,
515            target_encryption,
516            selection,
517            controls,
518            &formatter,
519            alias_manager,
520        )
521        .await;
522    }
523
524    let Some(source) = sources.first() else {
525        return formatter.fail(ExitCode::UsageError, "At least one copy source is required");
526    };
527
528    execute_single_copy(
529        source,
530        &target,
531        &args,
532        &formatter,
533        target_encryption.as_ref(),
534    )
535    .await
536}
537
538fn validate_alias_direction(
539    alias: TransferAlias,
540    sources: &[ParsedPath],
541    target: &ParsedPath,
542) -> Result<(), &'static str> {
543    match alias {
544        TransferAlias::Copy => Ok(()),
545        TransferAlias::Get if sources.len() == 1 && sources[0].is_remote() && target.is_local() => {
546            Ok(())
547        }
548        TransferAlias::Get => Err("get requires exactly one remote source and one local target"),
549        TransferAlias::Put if sources.iter().all(ParsedPath::is_local) && target.is_remote() => {
550            Ok(())
551        }
552        TransferAlias::Put => Err("put requires one or more local sources and one remote target"),
553    }
554}
555
556fn uses_transfer_planner(args: &CpArgs) -> bool {
557    args.sources.len() > 1
558        || args.recursive
559        || args.skip_existing
560        || !args.overwrite
561        || !args.include.is_empty()
562        || !args.exclude.is_empty()
563        || args.newer_than.is_some()
564        || args.older_than.is_some()
565        || args.rewind.is_some()
566        || args.concurrency.is_some()
567        || args.rate_limit.is_some()
568        || args.retry_attempts.is_some()
569        || args.retry_initial_backoff_ms.is_some()
570        || args.retry_max_backoff_ms.is_some()
571        || args.fail_empty
572        || args.summary
573}
574
575pub(super) fn validate_destination_storage_class(value: Option<&str>) -> rc_core::Result<()> {
576    let Some(value) = value else {
577        return Ok(());
578    };
579    match value {
580        "STANDARD" | "REDUCED_REDUNDANCY" => Ok(()),
581        "DEEP_ARCHIVE"
582        | "EXPRESS_ONEZONE"
583        | "FSX_ONTAP"
584        | "FSX_OPENZFS"
585        | "GLACIER"
586        | "GLACIER_IR"
587        | "INTELLIGENT_TIERING"
588        | "ONEZONE_IA"
589        | "OUTPOSTS"
590        | "SNOW"
591        | "STANDARD_IA" => Err(Error::UnsupportedFeature(format!(
592            "RustFS beta.10 does not provide meaningful storage policy '{value}'"
593        ))),
594        value => Err(Error::InvalidPath(format!(
595            "Unknown destination storage class '{value}'"
596        ))),
597    }
598}
599
600fn object_write_options(
601    fidelity: &TransferFidelityArgs,
602    content_type: Option<&str>,
603    encryption: Option<&ObjectEncryptionRequest>,
604    customer_key: Option<&SseCustomerKey>,
605    storage_class: Option<String>,
606) -> rc_core::Result<ObjectWriteOptions> {
607    fidelity.build_write_options(content_type, encryption, customer_key, storage_class)
608}
609
610fn requested_metadata_directive(args: &CpArgs) -> Option<MetadataDirective> {
611    if args.preserve {
612        Some(MetadataDirective::Copy)
613    } else {
614        args.metadata_directive.map(Into::into)
615    }
616}
617
618fn transfer_copy_options(
619    args: &CpArgs,
620    source_version_id: Option<String>,
621    encryption: Option<&ObjectEncryptionRequest>,
622) -> rc_core::Result<TransferCopyOptions> {
623    let metadata_directive = requested_metadata_directive(args);
624    let tagging_directive = args.tagging_directive.map(Into::into);
625    let mut destination = object_write_options(
626        &args.fidelity,
627        args.content_type.as_deref(),
628        encryption,
629        args.destination_customer_key.as_ref(),
630        args.storage_class.clone(),
631    )?;
632    if matches!(metadata_directive, Some(MetadataDirective::Replace))
633        && destination.attributes.is_none()
634    {
635        destination.attributes = Some(Default::default());
636    }
637    if matches!(tagging_directive, Some(rc_core::TaggingDirective::Replace))
638        && destination.tags.is_none()
639    {
640        destination.tags = Some(Default::default());
641    }
642    let options = TransferCopyOptions {
643        source: TransferReadOptions {
644            version_id: source_version_id,
645            customer_key: args.source_customer_key.clone(),
646            ..TransferReadOptions::default()
647        },
648        metadata_directive,
649        tagging_directive,
650        destination,
651    };
652    options.validate()?;
653    Ok(options)
654}
655
656fn validate_fidelity_directions(
657    args: &CpArgs,
658    sources: &[ParsedPath],
659    target: &ParsedPath,
660) -> rc_core::Result<()> {
661    let all_remote = sources
662        .iter()
663        .all(|source| matches!(source, ParsedPath::Remote(_)));
664    let any_remote = sources
665        .iter()
666        .any(|source| matches!(source, ParsedPath::Remote(_)));
667    let target_remote = matches!(target, ParsedPath::Remote(_));
668    let has_copy_directive =
669        args.preserve || args.metadata_directive.is_some() || args.tagging_directive.is_some();
670    if has_copy_directive && !(all_remote && target_remote) {
671        return Err(Error::InvalidPath(
672            "Metadata and tagging copy directives require remote sources and a remote destination"
673                .to_string(),
674        ));
675    }
676    if !target_remote
677        && (args.storage_class.is_some()
678            || args.content_type.is_some()
679            || args.fidelity.has_write_policy()
680            || !args.enc_s3.is_empty()
681            || !args.enc_kms.is_empty()
682            || args.enc_c_destination_key_file.is_some()
683            || args.enc_c_destination_key_env.is_some())
684    {
685        return Err(Error::InvalidPath(
686            "Destination transfer policies require a remote destination".to_string(),
687        ));
688    }
689    let same_alias_remote_copy = target_remote
690        && sources.iter().any(|source| {
691            matches!(
692                source,
693                ParsedPath::Remote(source)
694                    if target
695                        .as_remote()
696                        .is_some_and(|target| source.alias == target.alias)
697            )
698        });
699    if any_remote && target_remote {
700        let copy_options = transfer_copy_options(args, None, None)?;
701        if same_alias_remote_copy
702            && matches!(
703                copy_options.metadata_directive,
704                Some(MetadataDirective::Replace)
705            )
706        {
707            return Err(Error::UnsupportedFeature(
708                "RustFS beta.10 does not preserve complete metadata REPLACE semantics; tracked by rustfs/backlog#1463"
709                    .to_string(),
710            ));
711        }
712        if copy_options.tagging_directive.is_some() || copy_options.destination.tags.is_some() {
713            return Err(Error::UnsupportedFeature(
714                "RustFS beta.10 does not preserve CopyObject tagging directives; tracked by rustfs/backlog#1462"
715                    .to_string(),
716            ));
717        }
718        if copy_options.destination.checksum.is_some() {
719            return Err(Error::UnsupportedFeature(
720                "RustFS beta.10 does not preserve CopyObject checksum selection; tracked by rustfs/backlog#1466"
721                    .to_string(),
722            ));
723        }
724    }
725    Ok(())
726}
727
728fn load_customer_key(locator: Option<&SecretLocator>) -> rc_core::Result<Option<SseCustomerKey>> {
729    locator.map(SecretLocator::load_customer_key).transpose()
730}
731
732async fn execute_single_copy(
733    source: &ParsedPath,
734    target: &ParsedPath,
735    args: &CpArgs,
736    formatter: &Formatter,
737    encryption: Option<&ObjectEncryptionRequest>,
738) -> ExitCode {
739    // Determine copy direction.
740    match (source, target) {
741        (ParsedPath::Local(src), ParsedPath::Remote(dst)) => {
742            copy_local_to_s3_prepared(src, dst, args, formatter, encryption).await
743        }
744        (ParsedPath::Remote(src), ParsedPath::Local(dst)) => {
745            copy_s3_to_local(src, dst, args, formatter).await
746        }
747        (ParsedPath::Remote(src), ParsedPath::Remote(dst)) => {
748            if args.recursive {
749                return formatter.fail(
750                    ExitCode::UnsupportedFeature,
751                    "Recursive S3-to-S3 copy is not implemented",
752                );
753            }
754            copy_s3_to_s3_prepared(src, dst, args, formatter, encryption).await
755        }
756        (ParsedPath::Local(_), ParsedPath::Local(_)) => formatter.fail_with_suggestion(
757            ExitCode::UsageError,
758            "Cannot copy between two local paths. Use system cp command.",
759            "Use your local shell cp command when both paths are on the filesystem.",
760        ),
761    }
762}
763
764#[derive(Debug, Clone)]
765enum CpOperation {
766    LocalToRemote {
767        source: PathBuf,
768        target: RemotePath,
769        encryption: Option<ObjectEncryptionRequest>,
770    },
771    RemoteToLocal {
772        source: RemotePath,
773        target: PathBuf,
774    },
775    RemoteToRemote {
776        source: RemotePath,
777        target: RemotePath,
778        source_info: Box<ObjectInfo>,
779        encryption: Option<ObjectEncryptionRequest>,
780    },
781}
782
783#[derive(Debug, Clone)]
784struct PlannedCopyDetail {
785    source_version_id: Option<String>,
786    destination_version_id: Option<String>,
787    upload_id: Option<String>,
788}
789
790type PlannedCopyDetails = Arc<AsyncMutex<HashMap<(String, String), PlannedCopyDetail>>>;
791
792#[derive(Debug)]
793struct PlannedCopyProgress {
794    positions: StdMutex<HashMap<(String, String), u64>>,
795    bar: Option<ProgressBar>,
796}
797
798impl PlannedCopyProgress {
799    fn new(output_config: OutputConfig, total: u64) -> Self {
800        Self {
801            positions: StdMutex::new(HashMap::new()),
802            bar: (total > 0).then(|| ProgressBar::new(output_config, total)),
803        }
804    }
805
806    fn reset(&self, key: &(String, String)) {
807        self.set(key, 0);
808    }
809
810    fn set(&self, key: &(String, String), bytes: u64) {
811        let mut positions = self
812            .positions
813            .lock()
814            .expect("planned copy progress lock should not be poisoned");
815        positions.insert(key.clone(), bytes);
816        let aggregate = positions.values().copied().fold(0_u64, u64::saturating_add);
817        if let Some(bar) = &self.bar {
818            bar.set_position(aggregate);
819        }
820    }
821
822    fn finish(&self) {
823        if let Some(bar) = &self.bar {
824            bar.finish_and_clear();
825        }
826    }
827}
828
829type SharedPlannedCopyProgress = Arc<PlannedCopyProgress>;
830
831#[allow(clippy::too_many_arguments)]
832async fn execute_transfer_plan(
833    args: &CpArgs,
834    sources: Vec<ParsedPath>,
835    target: ParsedPath,
836    target_is_container: bool,
837    encryption: Option<ObjectEncryptionRequest>,
838    selection: TransferSelection,
839    controls: TransferControls,
840    formatter: &Formatter,
841    alias_manager: &AliasManager,
842) -> ExitCode {
843    let candidates = match build_transfer_candidates(
844        &sources,
845        &target,
846        target_is_container,
847        args.recursive,
848        encryption,
849        args.source_customer_key.as_ref(),
850        alias_manager,
851        args.local_key_policy(),
852    )
853    .await
854    {
855        Ok(candidates) => candidates,
856        Err(error) => {
857            return formatter.fail(
858                exit_code_for_core_error(&error),
859                &format!("Failed to plan copy: {error}"),
860            );
861        }
862    };
863
864    let mut plan = TransferPlan::build(candidates, &selection);
865    if let Err(error) = validate_storage_class_plan(&plan, args.storage_class.as_deref()) {
866        return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
867    }
868    if let Err(error) = validate_plan_targets(&plan) {
869        return formatter.fail(ExitCode::UsageError, &error.to_string());
870    }
871    if plan.items.is_empty() {
872        if args.fail_empty {
873            return formatter.fail(
874                ExitCode::NotFound,
875                "No copy sources matched the requested selection",
876            );
877        }
878        if args.summary || args.recursive || args.sources.len() > 1 {
879            print_transfer_summary(formatter, &plan.summary);
880        }
881        return ExitCode::Success;
882    }
883
884    let clients = match create_planned_client_cache(&plan.items, alias_manager).await {
885        Ok(clients) => Arc::new(clients),
886        Err(error) => {
887            return formatter.fail(
888                exit_code_for_core_error(&error),
889                &format!("Failed to prepare copy clients: {error}"),
890            );
891        }
892    };
893    let skipped_existing = if args.skip_existing || !args.overwrite {
894        match skip_existing_remote_targets(
895            &mut plan,
896            &clients,
897            args.destination_customer_key.as_ref(),
898        )
899        .await
900        {
901            Ok(skipped) => skipped,
902            Err(error) => {
903                return formatter.fail(
904                    exit_code_for_core_error(&error),
905                    &format!("Failed to inspect copy destinations: {error}"),
906                );
907            }
908        }
909    } else {
910        Vec::new()
911    };
912
913    if args.dry_run {
914        for item in &plan.items {
915            formatter.println(&format!(
916                "Would copy: {} -> {}{}",
917                formatter.style_file(&item.source),
918                formatter.style_file(&item.target),
919                transfer_policy_suffix(args)
920            ));
921        }
922        for item in &skipped_existing {
923            print_skipped_existing(formatter, item, true);
924        }
925        if args.summary || args.recursive || args.sources.len() > 1 {
926            print_transfer_summary(formatter, &plan.summary);
927        }
928        return ExitCode::Success;
929    }
930
931    for item in &skipped_existing {
932        print_skipped_existing(formatter, item, false);
933    }
934    if plan.items.is_empty() {
935        if args.summary || args.recursive || args.sources.len() > 1 {
936            print_transfer_summary(formatter, &plan.summary);
937        }
938        return ExitCode::Success;
939    }
940
941    let total_bytes = plan
942        .items
943        .iter()
944        .filter_map(|item| {
945            if matches!(item.payload, CpOperation::RemoteToRemote { .. }) {
946                item.size_bytes
947            } else {
948                None
949            }
950        })
951        .sum::<u64>();
952    let executor = match TransferExecutor::new(controls) {
953        Ok(executor) => executor,
954        Err(error) => {
955            return formatter.fail(ExitCode::UsageError, &error.to_string());
956        }
957    };
958    let operation_args = Arc::new(args.clone());
959    let transfer_cancellation = TransferCancellation::new();
960    let multipart_cancellation = MultipartCopyCancellation::new();
961    let signal_task = tokio::spawn({
962        let transfer_cancellation = transfer_cancellation.clone();
963        let multipart_cancellation = multipart_cancellation.clone();
964        async move {
965            if tokio::signal::ctrl_c().await.is_ok() {
966                multipart_cancellation.cancel();
967                transfer_cancellation.cancel();
968            }
969        }
970    });
971    let copy_details = PlannedCopyDetails::default();
972    let copy_progress = Arc::new(PlannedCopyProgress::new(
973        formatter.output_config(),
974        total_bytes,
975    ));
976    let report = executor
977        .execute_with_cancellation(plan, transfer_cancellation, {
978            let operation_args = Arc::clone(&operation_args);
979            let clients = Arc::clone(&clients);
980            let multipart_cancellation = multipart_cancellation.clone();
981            let copy_details = Arc::clone(&copy_details);
982            let copy_progress = Arc::clone(&copy_progress);
983            move |item| {
984                let operation_args = Arc::clone(&operation_args);
985                let clients = Arc::clone(&clients);
986                let multipart_cancellation = multipart_cancellation.clone();
987                let copy_details = Arc::clone(&copy_details);
988                let copy_progress = Arc::clone(&copy_progress);
989                async move {
990                    execute_planned_operation(
991                        item,
992                        &operation_args,
993                        &clients,
994                        &multipart_cancellation,
995                        &copy_details,
996                        &copy_progress,
997                    )
998                    .await
999                }
1000            }
1001        })
1002        .await;
1003    signal_task.abort();
1004    let _ = signal_task.await;
1005    copy_progress.finish();
1006
1007    let copy_details = copy_details.lock().await;
1008    for outcome in &report.outcomes {
1009        match &outcome.state {
1010            TransferOutcomeState::Success { bytes_transferred } => {
1011                let key = (outcome.item.source.clone(), outcome.item.target.clone());
1012                print_planned_success(
1013                    formatter,
1014                    &outcome.item,
1015                    *bytes_transferred,
1016                    copy_details.get(&key),
1017                );
1018            }
1019            TransferOutcomeState::Failed { error } => {
1020                formatter.error_with_code(exit_code_for_core_error(error), &error.to_string());
1021            }
1022            TransferOutcomeState::Cancelled { error } => {
1023                if let Some(error) = error {
1024                    formatter.warning(&format!(
1025                        "Cancelled transfer: {} ({error})",
1026                        outcome.item.source
1027                    ));
1028                } else {
1029                    formatter.warning(&format!(
1030                        "Cancelled before transfer: {}",
1031                        outcome.item.source
1032                    ));
1033                }
1034            }
1035        }
1036    }
1037
1038    if args.summary || args.recursive || args.sources.len() > 1 {
1039        print_transfer_summary(formatter, &report.summary);
1040    }
1041
1042    if report.was_cancelled {
1043        ExitCode::Interrupted
1044    } else {
1045        report
1046            .first_failure()
1047            .map_or(ExitCode::Success, exit_code_for_core_error)
1048    }
1049}
1050
1051fn transfer_policy_suffix(args: &CpArgs) -> String {
1052    let mut policies = Vec::new();
1053    if let Some(value) = args.storage_class.as_deref() {
1054        policies.push(format!("storage-class={value}"));
1055    }
1056    if let Some(directive) = requested_metadata_directive(args) {
1057        policies.push(format!(
1058            "metadata={}",
1059            match directive {
1060                MetadataDirective::Copy => "copy",
1061                MetadataDirective::Replace => "replace",
1062            }
1063        ));
1064    } else if args.fidelity.has_attribute_policy() || args.content_type.is_some() {
1065        policies.push(format!("metadata=write({})", args.fidelity.metadata.len()));
1066    }
1067    if let Some(directive) = args.tagging_directive {
1068        policies.push(format!(
1069            "tags={}({})",
1070            match directive {
1071                TaggingDirectiveArg::Copy => "copy",
1072                TaggingDirectiveArg::Replace => "replace",
1073            },
1074            args.fidelity.tags.len()
1075        ));
1076    } else if args.fidelity.has_tag_policy() {
1077        policies.push(format!("tags=write({})", args.fidelity.tags.len()));
1078    }
1079    if args.fidelity.checksum.is_some() {
1080        policies.push("checksum=sha256".to_string());
1081    }
1082    if let Some(mode) = args.fidelity.retention_mode.as_deref() {
1083        policies.push(format!("retention={}", mode.to_ascii_lowercase()));
1084    }
1085    if let Some(state) = args.fidelity.legal_hold.as_deref() {
1086        policies.push(format!("legal-hold={}", state.to_ascii_lowercase()));
1087    }
1088    if args.enc_c_destination_key_file.is_some() || args.enc_c_destination_key_env.is_some() {
1089        policies.push("encryption=sse-c".to_string());
1090    } else if !args.enc_kms.is_empty() {
1091        policies.push("encryption=sse-kms".to_string());
1092    } else if !args.enc_s3.is_empty() {
1093        policies.push("encryption=sse-s3".to_string());
1094    }
1095    if policies.is_empty() {
1096        String::new()
1097    } else {
1098        format!(" [policy:{}]", policies.join(","))
1099    }
1100}
1101
1102fn validate_storage_class_plan(
1103    plan: &TransferPlan<CpOperation>,
1104    storage_class: Option<&str>,
1105) -> rc_core::Result<()> {
1106    if storage_class.is_none() {
1107        return Ok(());
1108    }
1109    for item in &plan.items {
1110        let size = item.size_bytes.ok_or_else(|| {
1111            Error::UnsupportedFeature(format!(
1112                "Cannot guarantee storage class for a transfer with unknown size: {}",
1113                item.source
1114            ))
1115        })?;
1116        let multipart = match &item.payload {
1117            CpOperation::LocalToRemote { .. } => size > MULTIPART_THRESHOLD,
1118            CpOperation::RemoteToRemote { source, target, .. } if source.alias != target.alias => {
1119                size > MULTIPART_THRESHOLD
1120            }
1121            CpOperation::RemoteToRemote { .. } => rc_core::requires_multipart_copy(size),
1122            CpOperation::RemoteToLocal { .. } => false,
1123        };
1124        if multipart {
1125            return Err(Error::UnsupportedFeature(format!(
1126                "RustFS beta.10 does not persist storage class for multipart transfer: {}",
1127                item.source
1128            )));
1129        }
1130    }
1131    Ok(())
1132}
1133
1134async fn execute_planned_operation(
1135    item: TransferCandidate<CpOperation>,
1136    args: &CpArgs,
1137    clients: &HashMap<String, Arc<S3Client>>,
1138    multipart_cancellation: &MultipartCopyCancellation,
1139    copy_details: &PlannedCopyDetails,
1140    copy_progress: &SharedPlannedCopyProgress,
1141) -> rc_core::Result<u64> {
1142    let client = planned_client(clients, operation_alias(&item.payload))?;
1143    match &item.payload {
1144        CpOperation::LocalToRemote {
1145            source,
1146            target,
1147            encryption,
1148        } => perform_planned_upload(client, source, target, encryption.as_ref(), args).await,
1149        CpOperation::RemoteToLocal { source, target } => {
1150            perform_planned_download(client, source, target, args).await
1151        }
1152        CpOperation::RemoteToRemote {
1153            source,
1154            target,
1155            source_info,
1156            encryption,
1157        } => {
1158            let source_client = planned_client(clients, &source.alias)?;
1159            let target_client = planned_client(clients, &target.alias)?;
1160            let progress_key = (item.source.clone(), item.target.clone());
1161            copy_progress.reset(&progress_key);
1162            let result = perform_planned_remote_copy(
1163                source_client,
1164                target_client,
1165                source,
1166                target,
1167                source_info,
1168                encryption.as_ref(),
1169                multipart_cancellation,
1170                &|bytes| copy_progress.set(&progress_key, bytes),
1171                args,
1172            )
1173            .await?;
1174            copy_progress.set(&progress_key, result.bytes_copied);
1175            copy_details.lock().await.insert(
1176                (item.source, item.target),
1177                PlannedCopyDetail {
1178                    source_version_id: result.source_version_id,
1179                    destination_version_id: result.destination_version_id,
1180                    upload_id: result.upload_id,
1181                },
1182            );
1183            Ok(result.bytes_copied)
1184        }
1185    }
1186}
1187
1188async fn perform_planned_upload(
1189    client: &S3Client,
1190    source: &Path,
1191    target: &RemotePath,
1192    encryption: Option<&ObjectEncryptionRequest>,
1193    args: &CpArgs,
1194) -> rc_core::Result<u64> {
1195    let metadata = tokio::fs::metadata(source).await?;
1196    let file_size = metadata.len();
1197    let guessed_type = mime_guess::from_path(source)
1198        .first()
1199        .map(|mime| mime.essence_str().to_string());
1200    let content_type = select_upload_content_type(
1201        args.content_type.as_deref(),
1202        guessed_type.as_deref(),
1203        file_size,
1204    );
1205    let options = object_write_options(
1206        &args.fidelity,
1207        content_type,
1208        encryption,
1209        args.destination_customer_key.as_ref(),
1210        args.storage_class.clone(),
1211    )?;
1212    let info = client
1213        .put_object_from_path_with_options(target, source, &options, |_| {})
1214        .await?;
1215    Ok(info
1216        .size_bytes
1217        .and_then(|size| u64::try_from(size).ok())
1218        .unwrap_or(file_size))
1219}
1220
1221async fn perform_planned_download(
1222    client: &S3Client,
1223    source: &RemotePath,
1224    target: &Path,
1225    args: &CpArgs,
1226) -> rc_core::Result<u64> {
1227    if target.exists() && !args.overwrite {
1228        return Err(Error::Conflict(format!(
1229            "Destination exists: {}. Use --overwrite to replace.",
1230            target.display()
1231        )));
1232    }
1233    if let Some(parent) = target.parent() {
1234        tokio::fs::create_dir_all(parent).await?;
1235    }
1236    client
1237        .download_object_to_path_with_transfer_options(
1238            source,
1239            target,
1240            &TransferReadOptions {
1241                customer_key: args.source_customer_key.clone(),
1242                ..TransferReadOptions::default()
1243            },
1244            |_, _| {},
1245        )
1246        .await
1247}
1248
1249struct PlannedRemoteCopyResult {
1250    bytes_copied: u64,
1251    source_version_id: Option<String>,
1252    source_etag: Option<String>,
1253    destination_version_id: Option<String>,
1254    upload_id: Option<String>,
1255    object: ObjectInfo,
1256}
1257
1258#[allow(clippy::too_many_arguments)]
1259async fn perform_planned_remote_copy(
1260    source_client: &S3Client,
1261    target_client: &S3Client,
1262    source: &RemotePath,
1263    target: &RemotePath,
1264    source_info: &ObjectInfo,
1265    encryption: Option<&ObjectEncryptionRequest>,
1266    cancellation: &MultipartCopyCancellation,
1267    on_progress: &(dyn Fn(u64) + Send + Sync),
1268    args: &CpArgs,
1269) -> rc_core::Result<PlannedRemoteCopyResult> {
1270    if source.alias != target.alias {
1271        return perform_cross_alias_remote_copy(
1272            source_client,
1273            target_client,
1274            source,
1275            target,
1276            source_info,
1277            encryption,
1278            on_progress,
1279            args,
1280        )
1281        .await;
1282    }
1283    if args.source_customer_key.is_some() || args.destination_customer_key.is_some() {
1284        return Err(Error::UnsupportedFeature(
1285            "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467"
1286                .to_string(),
1287        ));
1288    }
1289    let planned_size = source_info
1290        .size_bytes
1291        .and_then(|size| u64::try_from(size).ok())
1292        .ok_or_else(|| Error::InvalidPath(format!("Source size is unavailable: {source}")))?;
1293    if rc_core::requires_multipart_copy(planned_size) {
1294        if args.storage_class.is_some() {
1295            return Err(Error::UnsupportedFeature(
1296                "RustFS beta.10 does not persist storage class for multipart copies".to_string(),
1297            ));
1298        }
1299        let current = source_client.head_object(source).await?;
1300        if !source_identity_matches(source_info, &current) {
1301            return Err(Error::Conflict(format!(
1302                "Source changed after copy planning: {source}"
1303            )));
1304        }
1305        let options = multipart_options_from_source(&current)?;
1306        let transfer = transfer_copy_options(args, current.version_id.clone(), encryption)?;
1307        let copied = source_client
1308            .multipart_copy_with_transfer_options(
1309                source,
1310                target,
1311                &options,
1312                &transfer,
1313                cancellation,
1314                on_progress,
1315            )
1316            .await?;
1317        return Ok(PlannedRemoteCopyResult {
1318            bytes_copied: copied.bytes_copied,
1319            source_version_id: options.source_version_id,
1320            source_etag: current.etag.clone(),
1321            destination_version_id: copied.object.version_id.clone(),
1322            upload_id: Some(copied.upload_id),
1323            object: copied.object,
1324        });
1325    }
1326    let options = transfer_copy_options(args, source_info.version_id.clone(), encryption)?;
1327    let copied = source_client
1328        .copy_object_with_transfer_options(source, target, &options)
1329        .await?;
1330    let bytes_copied = copied
1331        .size_bytes
1332        .and_then(|size| u64::try_from(size).ok())
1333        .unwrap_or(planned_size);
1334    on_progress(bytes_copied);
1335    Ok(PlannedRemoteCopyResult {
1336        bytes_copied,
1337        source_version_id: copied
1338            .source_version_id
1339            .clone()
1340            .or_else(|| source_info.version_id.clone()),
1341        source_etag: source_info.etag.clone(),
1342        destination_version_id: copied.version_id.clone(),
1343        upload_id: None,
1344        object: copied,
1345    })
1346}
1347
1348#[allow(clippy::too_many_arguments)]
1349async fn perform_cross_alias_remote_copy(
1350    source_client: &S3Client,
1351    target_client: &S3Client,
1352    source: &RemotePath,
1353    target: &RemotePath,
1354    source_info: &ObjectInfo,
1355    encryption: Option<&ObjectEncryptionRequest>,
1356    on_progress: &(dyn Fn(u64) + Send + Sync),
1357    args: &CpArgs,
1358) -> rc_core::Result<PlannedRemoteCopyResult> {
1359    // Server-side CopyObject cannot target a different alias/endpoint. Stream
1360    // through a bounded temporary file so the destination write is a normal
1361    // upload with the destination alias credentials.
1362    if args.source_customer_key.is_some() || args.destination_customer_key.is_some() {
1363        return Err(Error::UnsupportedFeature(
1364            "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467"
1365                .to_string(),
1366        ));
1367    }
1368    let planned_size = source_info
1369        .size_bytes
1370        .and_then(|size| u64::try_from(size).ok())
1371        .ok_or_else(|| Error::InvalidPath(format!("Source size is unavailable: {source}")))?;
1372    if args.storage_class.is_some() && planned_size > MULTIPART_THRESHOLD {
1373        return Err(Error::UnsupportedFeature(
1374            "RustFS beta.10 does not persist storage class for multipart uploads".to_string(),
1375        ));
1376    }
1377
1378    let current = source_client.head_object(source).await?;
1379    if !source_identity_matches(source_info, &current) {
1380        return Err(Error::Conflict(format!(
1381            "Source changed after copy planning: {source}"
1382        )));
1383    }
1384
1385    let staging = tempfile::Builder::new()
1386        .prefix("rc-cp-cross-alias-")
1387        .suffix(".part")
1388        .tempfile()?
1389        .into_temp_path();
1390    async {
1391        let downloaded = source_client
1392            .download_object_to_path_with_transfer_options(
1393                source,
1394                &staging,
1395                &TransferReadOptions {
1396                    version_id: current.version_id.clone(),
1397                    customer_key: args.source_customer_key.clone(),
1398                    ..TransferReadOptions::default()
1399                },
1400                |copied, _total| on_progress(copied),
1401            )
1402            .await?;
1403        if downloaded != planned_size {
1404            return Err(Error::Conflict(format!(
1405                "Source changed after copy planning: {source}"
1406            )));
1407        }
1408        let after = source_client
1409            .head_object_with_transfer_options(
1410                source,
1411                &TransferReadOptions {
1412                    version_id: current.version_id.clone(),
1413                    customer_key: args.source_customer_key.clone(),
1414                    ..TransferReadOptions::default()
1415                },
1416            )
1417            .await?;
1418        if !source_identity_matches(&current, &after) {
1419            return Err(Error::Conflict(format!(
1420                "Source changed after copy planning: {source}"
1421            )));
1422        }
1423        let options = piped_copy_write_options(args, &current, encryption)?;
1424        let object = target_client
1425            .put_object_from_path_with_options(target, &staging, &options, |copied| {
1426                on_progress(copied)
1427            })
1428            .await?;
1429        let bytes_copied = object
1430            .size_bytes
1431            .and_then(|size| u64::try_from(size).ok())
1432            .unwrap_or(downloaded);
1433        on_progress(bytes_copied);
1434        Ok(PlannedRemoteCopyResult {
1435            bytes_copied,
1436            source_version_id: current.version_id.clone(),
1437            source_etag: current.etag.clone(),
1438            destination_version_id: object.version_id.clone(),
1439            upload_id: None,
1440            object,
1441        })
1442    }
1443    .await
1444}
1445
1446fn source_identity_matches(planned: &ObjectInfo, current: &ObjectInfo) -> bool {
1447    if planned.size_bytes != current.size_bytes {
1448        return false;
1449    }
1450    match (&planned.etag, &current.etag) {
1451        (Some(planned), Some(current)) => planned == current,
1452        // Without an ETag an unversioned object has no stable read identity;
1453        // only an identical explicit version can prove that it is unchanged.
1454        (None, None) => {
1455            planned.version_id.is_some()
1456                && planned.version_id.as_ref() == current.version_id.as_ref()
1457        }
1458        _ => false,
1459    }
1460}
1461
1462/// Copy one object between two aliases by streaming through the client.
1463///
1464/// `rc mv` shares this path so a cross-alias move behaves exactly like a
1465/// cross-alias copy followed by a source delete, rather than reimplementing the
1466/// download/upload streaming and its source-change checks.
1467#[derive(Debug, Clone)]
1468pub(super) struct CrossAliasCopyResult {
1469    pub(super) object: ObjectInfo,
1470    pub(super) source_version_id: Option<String>,
1471    pub(super) source_etag: Option<String>,
1472}
1473
1474pub(super) async fn copy_object_across_aliases(
1475    source_client: &S3Client,
1476    target_client: &S3Client,
1477    source: &RemotePath,
1478    target: &RemotePath,
1479    encryption: Option<&ObjectEncryptionRequest>,
1480) -> rc_core::Result<CrossAliasCopyResult> {
1481    let source_info = source_client.head_object(source).await?;
1482    let args = CpArgs::single(source.to_string(), target.to_string());
1483    let ignore_progress = |_: u64| {};
1484    let result = perform_cross_alias_remote_copy(
1485        source_client,
1486        target_client,
1487        source,
1488        target,
1489        &source_info,
1490        encryption,
1491        &ignore_progress,
1492        &args,
1493    )
1494    .await?;
1495    Ok(CrossAliasCopyResult {
1496        object: result.object,
1497        source_version_id: result.source_version_id,
1498        source_etag: result.source_etag,
1499    })
1500}
1501
1502fn piped_copy_write_options(
1503    args: &CpArgs,
1504    source: &ObjectInfo,
1505    encryption: Option<&ObjectEncryptionRequest>,
1506) -> rc_core::Result<ObjectWriteOptions> {
1507    let mut options = object_write_options(
1508        &args.fidelity,
1509        args.content_type.as_deref(),
1510        encryption,
1511        args.destination_customer_key.as_ref(),
1512        args.storage_class.clone(),
1513    )?;
1514    let replace_metadata = matches!(
1515        requested_metadata_directive(args),
1516        Some(MetadataDirective::Replace)
1517    );
1518    let mut attributes = options.attributes.take().unwrap_or_default();
1519    if !replace_metadata {
1520        if attributes.content_type.is_none() {
1521            attributes.content_type = source.content_type.clone();
1522        }
1523        if attributes.user_metadata.is_empty()
1524            && let Some(metadata) = &source.metadata
1525        {
1526            attributes.user_metadata.clone_from(metadata);
1527        }
1528    }
1529    // The destination computes its own ETag, so record the source ETag the same
1530    // way `rc mirror` does. Without this a later `mirror --compare auto` cannot
1531    // tell a faithful cross-alias copy from a changed object and recopies it.
1532    // This is `rc` bookkeeping rather than user data, so it survives --metadata-directive replace.
1533    if let Some(source_etag) = source.etag.as_deref() {
1534        set_source_identity(&mut attributes, source_etag);
1535    }
1536    if attributes != ObjectAttributes::default() {
1537        options.attributes = Some(attributes);
1538    }
1539    Ok(options)
1540}
1541
1542#[cfg(test)]
1543fn requires_multipart_copy(planned_size: Option<u64>) -> bool {
1544    planned_size.is_some_and(rc_core::requires_multipart_copy)
1545}
1546
1547fn multipart_options_from_source(source: &ObjectInfo) -> rc_core::Result<MultipartCopyOptions> {
1548    let source_size = source
1549        .size_bytes
1550        .and_then(|size| u64::try_from(size).ok())
1551        .ok_or_else(|| {
1552            Error::InvalidPath("Multipart copy source size is unavailable".to_string())
1553        })?;
1554    let source_etag = source.etag.clone().ok_or_else(|| {
1555        Error::InvalidPath("Multipart copy source ETag is unavailable".to_string())
1556    })?;
1557    let mut options = MultipartCopyOptions::new(source_size, source_etag)?;
1558    options.source_version_id = source.version_id.clone();
1559    options.content_type = source.content_type.clone();
1560    let mut attributes = ObjectAttributes {
1561        user_metadata: source.metadata.clone().unwrap_or_default(),
1562        ..ObjectAttributes::default()
1563    };
1564    if let Some(source_etag) = source.etag.as_deref() {
1565        set_source_identity(&mut attributes, source_etag);
1566    }
1567    options.metadata = attributes.user_metadata;
1568    Ok(options)
1569}
1570
1571fn operation_alias(operation: &CpOperation) -> &str {
1572    match operation {
1573        CpOperation::LocalToRemote { target, .. } => &target.alias,
1574        CpOperation::RemoteToLocal { source, .. } | CpOperation::RemoteToRemote { source, .. } => {
1575            &source.alias
1576        }
1577    }
1578}
1579
1580fn operation_client_aliases(operation: &CpOperation) -> Vec<&str> {
1581    match operation {
1582        CpOperation::LocalToRemote { target, .. } => vec![target.alias.as_str()],
1583        CpOperation::RemoteToLocal { source, .. } => vec![source.alias.as_str()],
1584        CpOperation::RemoteToRemote { source, target, .. } => {
1585            if source.alias == target.alias {
1586                vec![source.alias.as_str()]
1587            } else {
1588                vec![source.alias.as_str(), target.alias.as_str()]
1589            }
1590        }
1591    }
1592}
1593
1594fn planned_client_aliases(items: &[TransferCandidate<CpOperation>]) -> BTreeSet<String> {
1595    items
1596        .iter()
1597        .flat_map(|item| {
1598            operation_client_aliases(&item.payload)
1599                .into_iter()
1600                .map(ToOwned::to_owned)
1601        })
1602        .collect()
1603}
1604
1605async fn create_planned_client_cache(
1606    items: &[TransferCandidate<CpOperation>],
1607    alias_manager: &AliasManager,
1608) -> rc_core::Result<HashMap<String, Arc<S3Client>>> {
1609    let mut clients = HashMap::new();
1610    for alias_name in planned_client_aliases(items) {
1611        match create_leaf_s3_client(alias_manager, &alias_name).await {
1612            Ok(client) => {
1613                clients.insert(alias_name, Arc::new(client));
1614            }
1615            // Keep a missing alias as an item-level failure so continue-on-error and summaries
1616            // retain their existing behavior without reloading configuration in each worker.
1617            Err(Error::AliasNotFound(_)) => {}
1618            Err(error) => return Err(error),
1619        }
1620    }
1621    Ok(clients)
1622}
1623
1624fn planned_client<'a>(
1625    clients: &'a HashMap<String, Arc<S3Client>>,
1626    alias_name: &str,
1627) -> rc_core::Result<&'a S3Client> {
1628    clients
1629        .get(alias_name)
1630        .map(Arc::as_ref)
1631        .ok_or_else(|| Error::AliasNotFound(alias_name.to_string()))
1632}
1633
1634async fn create_leaf_s3_client(
1635    alias_manager: &AliasManager,
1636    alias_name: &str,
1637) -> rc_core::Result<S3Client> {
1638    let mut alias = alias_manager
1639        .get(alias_name)
1640        .map_err(|_| Error::AliasNotFound(alias_name.to_string()))?;
1641    // The shared executor classifies failures and owns the exact attempt budget. Disabling adapter
1642    // retries here prevents nested retries from exceeding the command-level policy.
1643    alias.retry = Some(RetryConfig {
1644        max_attempts: 1,
1645        initial_backoff_ms: 1,
1646        max_backoff_ms: 1,
1647    });
1648    S3Client::new(alias).await
1649}
1650
1651fn print_planned_success(
1652    formatter: &Formatter,
1653    item: &TransferCandidate<CpOperation>,
1654    bytes_transferred: u64,
1655    detail: Option<&PlannedCopyDetail>,
1656) {
1657    let size_bytes = i64::try_from(bytes_transferred).ok();
1658    let size_human = size_bytes.map(|size| humansize::format_size(size as u64, humansize::BINARY));
1659    if formatter.is_json() {
1660        formatter.json(&CpOutput {
1661            status: "success",
1662            source: item.source.clone(),
1663            target: item.target.clone(),
1664            size_bytes,
1665            size_human,
1666            version_id: None,
1667            source_version_id: None,
1668        });
1669    } else {
1670        let mut line = format!(
1671            "{} -> {} ({})",
1672            formatter.style_file(&item.source),
1673            formatter.style_file(&item.target),
1674            formatter.style_size(size_human.as_deref().unwrap_or_default())
1675        );
1676        if let Some(detail) = detail {
1677            if let Some(version_id) = &detail.source_version_id {
1678                line.push_str(&format!(" source-version={version_id}"));
1679            }
1680            if let Some(version_id) = &detail.destination_version_id {
1681                line.push_str(&format!(" destination-version={version_id}"));
1682            }
1683            if let Some(upload_id) = &detail.upload_id {
1684                line.push_str(&format!(" upload-id={upload_id}"));
1685            }
1686        }
1687        formatter.println(&line);
1688    }
1689}
1690
1691fn print_skipped_existing(
1692    formatter: &Formatter,
1693    item: &TransferCandidate<CpOperation>,
1694    dry_run: bool,
1695) {
1696    let action = if dry_run {
1697        "Would skip existing"
1698    } else {
1699        "Skipped existing"
1700    };
1701    formatter.println(&format!(
1702        "{action}: {} -> {}",
1703        formatter.style_file(&item.source),
1704        formatter.style_file(&item.target)
1705    ));
1706}
1707
1708fn print_transfer_summary(formatter: &Formatter, summary: &rc_core::TransferSummary) {
1709    formatter.println(&format!(
1710        "Summary: {} planned, {} skipped, {} succeeded, {} failed, {} cancelled, {} transferred",
1711        summary.planned,
1712        summary.skipped,
1713        summary.successful,
1714        summary.failed,
1715        summary.cancelled,
1716        humansize::format_size(summary.transferred_bytes, humansize::BINARY)
1717    ));
1718}
1719
1720pub(super) fn exit_code_for_core_error(error: &Error) -> ExitCode {
1721    ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError)
1722}
1723
1724fn build_transfer_selection(args: &CpArgs, now: Timestamp) -> Result<TransferSelection, String> {
1725    let newer_than = args
1726        .newer_than
1727        .as_deref()
1728        .map(|value| parse_age_cutoff(value, now))
1729        .transpose()?;
1730    let older_than = args
1731        .older_than
1732        .as_deref()
1733        .map(|value| parse_age_cutoff(value, now))
1734        .transpose()?;
1735    let rewind = args
1736        .rewind
1737        .as_deref()
1738        .map(|value| parse_rewind_cutoff(value, now))
1739        .transpose()?;
1740
1741    TransferSelection::new(&args.include, &args.exclude, newer_than, older_than, rewind)
1742        .map_err(|error| error.to_string())
1743}
1744
1745fn build_transfer_controls(args: &CpArgs) -> Result<TransferControls, String> {
1746    let bytes_per_second = args
1747        .rate_limit
1748        .as_deref()
1749        .map(parse_byte_rate)
1750        .transpose()?;
1751    let controls = TransferControls {
1752        concurrency: args.concurrency.unwrap_or(DEFAULT_TRANSFER_CONCURRENCY),
1753        bytes_per_second,
1754        retry: RetryConfig {
1755            max_attempts: args.retry_attempts.unwrap_or(DEFAULT_RETRY_ATTEMPTS),
1756            initial_backoff_ms: args
1757                .retry_initial_backoff_ms
1758                .unwrap_or(DEFAULT_RETRY_INITIAL_BACKOFF_MS),
1759            max_backoff_ms: args
1760                .retry_max_backoff_ms
1761                .unwrap_or(DEFAULT_RETRY_MAX_BACKOFF_MS),
1762        },
1763        continue_on_error: args.continue_on_error,
1764    };
1765    controls.validate().map_err(|error| error.to_string())?;
1766    Ok(controls)
1767}
1768
1769pub(super) fn parse_age_cutoff(value: &str, now: Timestamp) -> Result<Timestamp, String> {
1770    let value = value.trim();
1771    if value.is_empty() {
1772        return Err("Transfer age must not be empty".to_string());
1773    }
1774    let suffix_index = value
1775        .find(|character: char| character.is_ascii_alphabetic())
1776        .unwrap_or(value.len());
1777    let number = &value[..suffix_index];
1778    let suffix = &value[suffix_index..];
1779    let amount: i64 = number
1780        .parse()
1781        .map_err(|_| format!("Invalid transfer age number: {number}"))?;
1782    if amount < 0 {
1783        return Err("Transfer age must not be negative".to_string());
1784    }
1785    let multiplier = match suffix.to_ascii_lowercase().as_str() {
1786        "" | "s" => 1,
1787        "m" => 60,
1788        "h" => 3_600,
1789        "d" => 86_400,
1790        "w" => 604_800,
1791        _ => return Err(format!("Unknown transfer age suffix: {suffix}")),
1792    };
1793    let seconds = amount
1794        .checked_mul(multiplier)
1795        .ok_or_else(|| "Transfer age is too large".to_string())?;
1796    now.checked_sub(jiff::Span::new().seconds(seconds))
1797        .map_err(|error| format!("Transfer age overflow: {error}"))
1798}
1799
1800fn parse_rewind_cutoff(value: &str, now: Timestamp) -> Result<Timestamp, String> {
1801    value
1802        .parse::<Timestamp>()
1803        .or_else(|_| parse_age_cutoff(value, now))
1804        .map_err(|error| format!("Invalid rewind value '{value}': {error}"))
1805}
1806
1807pub(super) fn parse_byte_rate(value: &str) -> Result<u64, String> {
1808    let normalized = value.trim().to_ascii_lowercase();
1809    let normalized = normalized
1810        .strip_suffix("/s")
1811        .or_else(|| normalized.strip_suffix("ps"))
1812        .unwrap_or(&normalized);
1813    let suffix_index = normalized
1814        .find(|character: char| character.is_ascii_alphabetic())
1815        .unwrap_or(normalized.len());
1816    let number = &normalized[..suffix_index];
1817    let suffix = &normalized[suffix_index..];
1818    let amount: u64 = number
1819        .parse()
1820        .map_err(|_| format!("Invalid transfer rate number: {number}"))?;
1821    let multiplier = match suffix {
1822        "" | "b" => 1u64,
1823        "k" | "kb" => 1_000,
1824        "ki" | "kib" => 1_024,
1825        "m" | "mb" => 1_000_000,
1826        "mi" | "mib" => 1_048_576,
1827        "g" | "gb" => 1_000_000_000,
1828        "gi" | "gib" => 1_073_741_824,
1829        _ => return Err(format!("Unknown transfer rate suffix: {suffix}")),
1830    };
1831    let rate = amount
1832        .checked_mul(multiplier)
1833        .ok_or_else(|| "Transfer rate is too large".to_string())?;
1834    if rate == 0 {
1835        return Err("Transfer rate must be greater than zero".to_string());
1836    }
1837    Ok(rate)
1838}
1839
1840fn is_container_target(raw: &str, target: &ParsedPath) -> bool {
1841    match target {
1842        ParsedPath::Local(path) => path.is_dir() || raw.ends_with(['/', '\\']),
1843        ParsedPath::Remote(path) => path.key.is_empty() || raw.ends_with('/'),
1844    }
1845}
1846
1847#[allow(clippy::too_many_arguments)]
1848async fn build_transfer_candidates(
1849    sources: &[ParsedPath],
1850    target: &ParsedPath,
1851    target_is_container: bool,
1852    recursive: bool,
1853    encryption: Option<ObjectEncryptionRequest>,
1854    source_customer_key: Option<&SseCustomerKey>,
1855    alias_manager: &AliasManager,
1856    key_policy: ObjectKeyPolicy,
1857) -> rc_core::Result<Vec<TransferCandidate<CpOperation>>> {
1858    let mut candidates = Vec::new();
1859    let mut planning_clients = HashMap::new();
1860    let multiple_sources = sources.len() > 1;
1861
1862    for source in sources {
1863        match source {
1864            ParsedPath::Local(source) => build_local_candidates(
1865                source,
1866                target,
1867                target_is_container,
1868                recursive,
1869                multiple_sources,
1870                encryption.clone(),
1871                &mut candidates,
1872            )?,
1873            ParsedPath::Remote(source) => {
1874                build_remote_candidates(
1875                    source,
1876                    target,
1877                    target_is_container,
1878                    recursive,
1879                    multiple_sources,
1880                    encryption.clone(),
1881                    source_customer_key,
1882                    alias_manager,
1883                    &mut planning_clients,
1884                    &mut candidates,
1885                    key_policy,
1886                )
1887                .await?;
1888            }
1889        }
1890    }
1891
1892    Ok(candidates)
1893}
1894
1895fn validate_plan_targets(plan: &TransferPlan<CpOperation>) -> rc_core::Result<()> {
1896    let mut targets = HashSet::with_capacity(plan.items.len());
1897    for candidate in &plan.items {
1898        if let CpOperation::RemoteToRemote { source, target, .. } = &candidate.payload
1899            && source == target
1900        {
1901            return Err(Error::InvalidPath(format!(
1902                "Source and destination resolve to the same object '{}'",
1903                candidate.source
1904            )));
1905        }
1906        if !targets.insert(candidate.target.clone()) {
1907            return Err(Error::InvalidPath(format!(
1908                "Multiple sources resolve to the same destination '{}'",
1909                candidate.target
1910            )));
1911        }
1912    }
1913    Ok(())
1914}
1915
1916async fn skip_existing_remote_targets(
1917    plan: &mut TransferPlan<CpOperation>,
1918    clients: &HashMap<String, Arc<S3Client>>,
1919    destination_customer_key: Option<&SseCustomerKey>,
1920) -> rc_core::Result<Vec<TransferCandidate<CpOperation>>> {
1921    let mut retained = Vec::with_capacity(plan.items.len());
1922    let mut skipped = Vec::new();
1923    for candidate in plan.items.drain(..) {
1924        let destination = match &candidate.payload {
1925            CpOperation::LocalToRemote { target, .. }
1926            | CpOperation::RemoteToRemote { target, .. } => Some(target),
1927            CpOperation::RemoteToLocal { .. } => None,
1928        };
1929        let Some(destination) = destination else {
1930            retained.push(candidate);
1931            continue;
1932        };
1933        let client = planned_client(clients, &destination.alias)?;
1934        match client
1935            .head_object_with_transfer_options(
1936                destination,
1937                &TransferReadOptions {
1938                    customer_key: destination_customer_key.cloned(),
1939                    ..TransferReadOptions::default()
1940                },
1941            )
1942            .await
1943        {
1944            Ok(_) => skipped.push(candidate),
1945            Err(Error::NotFound(_)) | Err(Error::VersionNotFound { .. }) => {
1946                retained.push(candidate);
1947            }
1948            Err(error) => return Err(error),
1949        }
1950    }
1951    plan.items = retained;
1952    plan.summary.planned = plan.items.len();
1953    plan.summary.skipped = plan.summary.skipped.saturating_add(skipped.len());
1954    Ok(skipped)
1955}
1956
1957#[allow(clippy::too_many_arguments)]
1958fn build_local_candidates(
1959    source: &Path,
1960    target: &ParsedPath,
1961    target_is_container: bool,
1962    recursive: bool,
1963    multiple_sources: bool,
1964    encryption: Option<ObjectEncryptionRequest>,
1965    candidates: &mut Vec<TransferCandidate<CpOperation>>,
1966) -> rc_core::Result<()> {
1967    let ParsedPath::Remote(target) = target else {
1968        return Err(Error::InvalidPath(
1969            "Cannot copy between two local paths. Use the system cp command.".to_string(),
1970        ));
1971    };
1972    let metadata = std::fs::metadata(source).map_err(|error| {
1973        if error.kind() == std::io::ErrorKind::NotFound {
1974            Error::NotFound(format!("Source not found: {}", source.display()))
1975        } else {
1976            Error::Io(error)
1977        }
1978    })?;
1979
1980    if metadata.is_file() {
1981        let name = local_file_name(source)?;
1982        let destination = if target_is_container {
1983            remote_child(target, &name, ObjectKeyPolicy::for_remote_destination())?
1984        } else {
1985            normalize_remote_target(target, ObjectKeyPolicy::for_remote_destination())?
1986        };
1987        candidates.push(local_transfer_candidate(
1988            source.to_path_buf(),
1989            destination,
1990            name,
1991            metadata,
1992            encryption,
1993        ));
1994        return Ok(());
1995    }
1996
1997    if !recursive {
1998        return Err(Error::InvalidPath(
1999            "Source is a directory. Use -r/--recursive to copy directories.".to_string(),
2000        ));
2001    }
2002    if !target_is_container {
2003        return Err(Error::InvalidPath(
2004            "Recursive copy requires a directory or remote prefix destination ending in '/'"
2005                .to_string(),
2006        ));
2007    }
2008
2009    let mut files = Vec::new();
2010    collect_local_files(source, source, &mut files)?;
2011    files.sort_by(|left, right| left.1.cmp(&right.1));
2012    let source_root = multiple_sources
2013        .then(|| local_file_name(source))
2014        .transpose()?;
2015    for (path, relative, metadata) in files {
2016        let relative = relative.replace('\\', "/");
2017        let target_relative = source_root
2018            .as_deref()
2019            .map_or_else(|| relative.clone(), |root| format!("{root}/{relative}"));
2020        candidates.push(local_transfer_candidate(
2021            path,
2022            remote_child(
2023                target,
2024                &target_relative,
2025                ObjectKeyPolicy::for_remote_destination(),
2026            )?,
2027            relative,
2028            metadata,
2029            encryption.clone(),
2030        ));
2031    }
2032    Ok(())
2033}
2034
2035fn local_file_name(path: &Path) -> rc_core::Result<String> {
2036    path.file_name()
2037        .map(|name| name.to_string_lossy().to_string())
2038        .filter(|name| !name.is_empty())
2039        .ok_or_else(|| Error::InvalidPath(format!("Path has no file name: {}", path.display())))
2040}
2041
2042fn local_transfer_candidate(
2043    source: PathBuf,
2044    target: RemotePath,
2045    relative_path: String,
2046    metadata: std::fs::Metadata,
2047    encryption: Option<ObjectEncryptionRequest>,
2048) -> TransferCandidate<CpOperation> {
2049    TransferCandidate {
2050        payload: CpOperation::LocalToRemote {
2051            source: source.clone(),
2052            target: target.clone(),
2053            encryption,
2054        },
2055        source: source.display().to_string(),
2056        target: target.to_string(),
2057        relative_path,
2058        modified: metadata
2059            .modified()
2060            .ok()
2061            .and_then(|time| time.try_into().ok()),
2062        size_bytes: Some(metadata.len()),
2063    }
2064}
2065
2066fn collect_local_files(
2067    directory: &Path,
2068    base: &Path,
2069    files: &mut Vec<(PathBuf, String, std::fs::Metadata)>,
2070) -> rc_core::Result<()> {
2071    let mut entries = std::fs::read_dir(directory)?.collect::<std::io::Result<Vec<_>>>()?;
2072    entries.sort_by_key(std::fs::DirEntry::file_name);
2073    for entry in entries {
2074        let path = entry.path();
2075        let file_type = entry.file_type()?;
2076        if file_type.is_symlink() {
2077            return Err(Error::InvalidPath(format!(
2078                "Symbolic links are not supported in recursive copy: {}",
2079                path.display()
2080            )));
2081        }
2082        let metadata = entry.metadata()?;
2083        if file_type.is_file() {
2084            let relative = path
2085                .strip_prefix(base)
2086                .map_err(|error| Error::InvalidPath(error.to_string()))?
2087                .to_string_lossy()
2088                .to_string();
2089            files.push((path, relative, metadata));
2090        } else if file_type.is_dir() {
2091            collect_local_files(&path, base, files)?;
2092        }
2093    }
2094    Ok(())
2095}
2096
2097#[allow(clippy::too_many_arguments)]
2098async fn build_remote_candidates(
2099    source: &RemotePath,
2100    target: &ParsedPath,
2101    target_is_container: bool,
2102    recursive: bool,
2103    multiple_sources: bool,
2104    encryption: Option<ObjectEncryptionRequest>,
2105    source_customer_key: Option<&SseCustomerKey>,
2106    alias_manager: &AliasManager,
2107    planning_clients: &mut HashMap<String, Arc<S3Client>>,
2108    candidates: &mut Vec<TransferCandidate<CpOperation>>,
2109    key_policy: ObjectKeyPolicy,
2110) -> rc_core::Result<()> {
2111    let is_prefix = source.key.is_empty() || source.key.ends_with('/') || recursive;
2112    let client = planning_client(planning_clients, alias_manager, &source.alias).await?;
2113
2114    if is_prefix {
2115        if !recursive {
2116            return Err(Error::InvalidPath(
2117                "Remote prefix copy requires -r/--recursive".to_string(),
2118            ));
2119        }
2120        if !target_is_container {
2121            return Err(Error::InvalidPath(
2122                "Recursive copy requires a directory or remote prefix destination ending in '/'"
2123                    .to_string(),
2124            ));
2125        }
2126
2127        let listing_source = recursive_listing_source(source);
2128        if let ParsedPath::Remote(target) = target
2129            && remote_copy_scopes_overlap(&listing_source, target)
2130        {
2131            return Err(Error::Conflict(format!(
2132                "Recursive source '{}' overlaps destination '{}'",
2133                listing_source, target
2134            )));
2135        }
2136
2137        let source_root = recursive_source_root(&listing_source, multiple_sources);
2138        let mut continuation_token = None;
2139        let mut seen_tokens = HashSet::new();
2140        loop {
2141            let result = client
2142                .list_objects(
2143                    &listing_source,
2144                    rc_core::ListOptions {
2145                        recursive: true,
2146                        max_keys: Some(1_000),
2147                        continuation_token: continuation_token.clone(),
2148                        ..Default::default()
2149                    },
2150                )
2151                .await?;
2152            for object in result.items.into_iter().filter(|object| !object.is_dir) {
2153                let object_source =
2154                    RemotePath::new(&listing_source.alias, &listing_source.bucket, &object.key);
2155                match target {
2156                    ParsedPath::Local(target_root) => {
2157                        let relative = safe_download_relative_path(
2158                            &object.key,
2159                            &listing_source.key,
2160                            key_policy,
2161                        )
2162                        .map_err(Error::InvalidPath)?;
2163                        let relative_string = relative.to_string_lossy().replace('\\', "/");
2164                        let target_relative = if source_root.is_empty() {
2165                            relative.clone()
2166                        } else {
2167                            PathBuf::from(&source_root).join(&relative)
2168                        };
2169                        let destination = safe_download_destination(target_root, &target_relative)
2170                            .await
2171                            .map_err(Error::InvalidPath)?;
2172                        candidates.push(TransferCandidate {
2173                            payload: CpOperation::RemoteToLocal {
2174                                source: object_source.clone(),
2175                                target: destination.clone(),
2176                            },
2177                            source: object_source.to_string(),
2178                            target: destination.display().to_string(),
2179                            relative_path: relative_string,
2180                            modified: object.last_modified,
2181                            size_bytes: object.size_bytes.and_then(|size| u64::try_from(size).ok()),
2182                        });
2183                    }
2184                    ParsedPath::Remote(target) => {
2185                        let (destination, relative) = recursive_remote_target(
2186                            &listing_source,
2187                            target,
2188                            &object.key,
2189                            multiple_sources,
2190                            ObjectKeyPolicy::for_remote_destination(),
2191                        )?;
2192                        let size_bytes =
2193                            object.size_bytes.and_then(|size| u64::try_from(size).ok());
2194                        candidates.push(TransferCandidate {
2195                            payload: CpOperation::RemoteToRemote {
2196                                source: object_source.clone(),
2197                                target: destination.clone(),
2198                                source_info: Box::new(object.clone()),
2199                                encryption: encryption.clone(),
2200                            },
2201                            source: object_source.to_string(),
2202                            target: destination.to_string(),
2203                            relative_path: relative,
2204                            modified: object.last_modified,
2205                            size_bytes,
2206                        });
2207                    }
2208                }
2209            }
2210            if !result.truncated {
2211                break;
2212            }
2213            let next_token = result.continuation_token.ok_or_else(|| {
2214                Error::InvalidPath(
2215                    "Truncated object listing did not include a continuation token".to_string(),
2216                )
2217            })?;
2218            if !seen_tokens.insert(next_token.clone()) {
2219                return Err(Error::Conflict(
2220                    "Object listing repeated a continuation token".to_string(),
2221                ));
2222            }
2223            continuation_token = Some(next_token);
2224        }
2225        return Ok(());
2226    }
2227
2228    let object = client
2229        .head_object_with_transfer_options(
2230            source,
2231            &TransferReadOptions {
2232                customer_key: source_customer_key.cloned(),
2233                ..TransferReadOptions::default()
2234            },
2235        )
2236        .await?;
2237    let name = source
2238        .key
2239        .rsplit('/')
2240        .next()
2241        .filter(|name| !name.is_empty())
2242        .ok_or_else(|| Error::InvalidPath(format!("Object key is empty: {source}")))?;
2243    let size_bytes = object.size_bytes.and_then(|size| u64::try_from(size).ok());
2244    let modified = object.last_modified;
2245
2246    match target {
2247        ParsedPath::Local(target) => {
2248            let destination = if target_is_container {
2249                let relative = safe_download_relative_path(name, "", key_policy)
2250                    .map_err(Error::InvalidPath)?;
2251                safe_download_destination(target, &relative)
2252                    .await
2253                    .map_err(Error::InvalidPath)?
2254            } else {
2255                target.clone()
2256            };
2257            candidates.push(TransferCandidate {
2258                payload: CpOperation::RemoteToLocal {
2259                    source: source.clone(),
2260                    target: destination.clone(),
2261                },
2262                source: source.to_string(),
2263                target: destination.display().to_string(),
2264                relative_path: name.to_string(),
2265                modified,
2266                size_bytes,
2267            });
2268        }
2269        ParsedPath::Remote(target) => {
2270            let destination = if target_is_container {
2271                remote_child(target, name, ObjectKeyPolicy::for_remote_destination())?
2272            } else {
2273                normalize_remote_target(target, ObjectKeyPolicy::for_remote_destination())?
2274            };
2275            candidates.push(TransferCandidate {
2276                payload: CpOperation::RemoteToRemote {
2277                    source: source.clone(),
2278                    target: destination.clone(),
2279                    source_info: Box::new(object),
2280                    encryption,
2281                },
2282                source: source.to_string(),
2283                target: destination.to_string(),
2284                relative_path: name.to_string(),
2285                modified,
2286                size_bytes,
2287            });
2288        }
2289    }
2290    Ok(())
2291}
2292
2293async fn planning_client(
2294    clients: &mut HashMap<String, Arc<S3Client>>,
2295    alias_manager: &AliasManager,
2296    alias_name: &str,
2297) -> rc_core::Result<Arc<S3Client>> {
2298    if let Some(client) = clients.get(alias_name) {
2299        return Ok(Arc::clone(client));
2300    }
2301    let alias = alias_manager
2302        .get(alias_name)
2303        .map_err(|_| Error::AliasNotFound(alias_name.to_string()))?;
2304    let client = Arc::new(S3Client::new(alias).await?);
2305    clients.insert(alias_name.to_string(), Arc::clone(&client));
2306    Ok(client)
2307}
2308
2309fn remote_child(
2310    parent: &RemotePath,
2311    relative: &str,
2312    policy: ObjectKeyPolicy,
2313) -> rc_core::Result<RemotePath> {
2314    let parent_key = normalize_remote_prefix(&parent.key, policy)?;
2315    let relative = normalize_relative_key(relative, policy)?;
2316    let key = if parent_key.is_empty() {
2317        relative
2318    } else {
2319        format!("{parent_key}/{relative}")
2320    };
2321    Ok(RemotePath::new(&parent.alias, &parent.bucket, key))
2322}
2323
2324fn normalize_remote_target(
2325    target: &RemotePath,
2326    policy: ObjectKeyPolicy,
2327) -> rc_core::Result<RemotePath> {
2328    let key = normalize_remote_prefix(&target.key, policy)?;
2329    if target.key.ends_with('/') && !key.is_empty() {
2330        Ok(RemotePath::new(
2331            &target.alias,
2332            &target.bucket,
2333            format!("{key}/"),
2334        ))
2335    } else {
2336        Ok(RemotePath::new(&target.alias, &target.bucket, key))
2337    }
2338}
2339
2340fn normalize_remote_prefix(key: &str, policy: ObjectKeyPolicy) -> rc_core::Result<String> {
2341    if key.is_empty() {
2342        return Ok(String::new());
2343    }
2344    normalize_relative_key(key, policy)
2345}
2346
2347fn recursive_listing_source(source: &RemotePath) -> RemotePath {
2348    let key = if source.key.is_empty() || source.key.ends_with('/') {
2349        source.key.clone()
2350    } else {
2351        format!("{}/", source.key)
2352    };
2353    RemotePath::new(&source.alias, &source.bucket, key)
2354}
2355
2356fn recursive_source_root(source: &RemotePath, multiple_sources: bool) -> String {
2357    if !multiple_sources {
2358        return String::new();
2359    }
2360    source
2361        .key
2362        .trim_end_matches('/')
2363        .rsplit('/')
2364        .next()
2365        .filter(|value| !value.is_empty())
2366        .unwrap_or(&source.bucket)
2367        .to_string()
2368}
2369
2370fn recursive_remote_target(
2371    source: &RemotePath,
2372    target: &RemotePath,
2373    object_key: &str,
2374    multiple_sources: bool,
2375    policy: ObjectKeyPolicy,
2376) -> rc_core::Result<(RemotePath, String)> {
2377    let relative = object_key.strip_prefix(&source.key).ok_or_else(|| {
2378        Error::InvalidPath(format!(
2379            "Listed object key '{object_key}' is outside source prefix '{}'",
2380            source.key
2381        ))
2382    })?;
2383    if relative.is_empty() {
2384        return Err(Error::InvalidPath(format!(
2385            "Listed object key '{object_key}' does not identify a child of source prefix '{}'",
2386            source.key
2387        )));
2388    }
2389    let destination_relative = match recursive_source_root(source, multiple_sources) {
2390        root if root.is_empty() => relative.to_string(),
2391        root => format!("{root}/{relative}"),
2392    };
2393    let normalized_relative = normalize_relative_key(relative, policy)?;
2394    let normalized_destination_relative = normalize_relative_key(&destination_relative, policy)?;
2395    Ok((
2396        remote_child(target, &normalized_destination_relative, policy)?,
2397        normalized_relative,
2398    ))
2399}
2400
2401fn remote_copy_scopes_overlap(source: &RemotePath, target: &RemotePath) -> bool {
2402    if source.alias != target.alias || source.bucket != target.bucket {
2403        return false;
2404    }
2405    let source_prefix = recursive_listing_source(source).key;
2406    let target_prefix = recursive_listing_source(target).key;
2407    source_prefix.is_empty()
2408        || target_prefix.is_empty()
2409        || source_prefix.starts_with(&target_prefix)
2410        || target_prefix.starts_with(&source_prefix)
2411}
2412
2413fn parse_cp_path(path: &str, alias_manager: Option<&AliasManager>) -> rc_core::Result<ParsedPath> {
2414    let parsed = parse_path(path)?;
2415
2416    let ParsedPath::Remote(remote) = &parsed else {
2417        return Ok(parsed);
2418    };
2419
2420    if let Some(manager) = alias_manager
2421        && matches!(manager.exists(&remote.alias), Ok(true))
2422    {
2423        return Ok(parsed);
2424    }
2425
2426    if Path::new(path).exists() {
2427        return Ok(ParsedPath::Local(PathBuf::from(path)));
2428    }
2429
2430    Ok(parsed)
2431}
2432
2433async fn copy_local_to_s3_prepared(
2434    src: &Path,
2435    dst: &RemotePath,
2436    args: &CpArgs,
2437    formatter: &Formatter,
2438    encryption: Option<&ObjectEncryptionRequest>,
2439) -> ExitCode {
2440    // Check if source exists
2441    if !src.exists() {
2442        return formatter.fail_with_suggestion(
2443            ExitCode::NotFound,
2444            &format!("Source not found: {}", src.display()),
2445            "Check the local source path and retry the copy command.",
2446        );
2447    }
2448
2449    // If source is a directory, require recursive flag
2450    if src.is_dir() && !args.recursive {
2451        return formatter.fail_with_suggestion(
2452            ExitCode::UsageError,
2453            "Source is a directory. Use -r/--recursive to copy directories.",
2454            "Retry with -r or --recursive to copy a directory tree.",
2455        );
2456    }
2457
2458    // Load alias and create client
2459    let alias_manager = match AliasManager::new() {
2460        Ok(am) => am,
2461        Err(e) => {
2462            formatter.error(&format!("Failed to load aliases: {e}"));
2463            return ExitCode::GeneralError;
2464        }
2465    };
2466
2467    let alias = match alias_manager.get(&dst.alias) {
2468        Ok(a) => a,
2469        Err(_) => {
2470            return formatter.fail_with_suggestion(
2471                ExitCode::NotFound,
2472                &format!("Alias '{}' not found", dst.alias),
2473                "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
2474            );
2475        }
2476    };
2477    let client = match S3Client::new(alias).await {
2478        Ok(c) => c,
2479        Err(e) => {
2480            return formatter.fail(
2481                ExitCode::NetworkError,
2482                &format!("Failed to create S3 client: {e}"),
2483            );
2484        }
2485    };
2486
2487    if src.is_file() {
2488        // Single file upload
2489        upload_file(&client, src, dst, args, formatter, encryption).await
2490    } else {
2491        // Directory upload
2492        upload_directory(&client, src, dst, args, formatter, encryption).await
2493    }
2494}
2495
2496/// Multipart upload threshold: files larger than this size use multipart upload.
2497const MULTIPART_THRESHOLD: u64 = rc_s3::multipart::DEFAULT_PART_SIZE;
2498/// Download progress threshold: avoid flicker for tiny downloads while surfacing meaningful waits.
2499const DOWNLOAD_PROGRESS_THRESHOLD: u64 = 4 * 1024 * 1024;
2500
2501fn update_download_progress(
2502    progress: &mut Option<ProgressBar>,
2503    output_config: &OutputConfig,
2504    bytes_downloaded: u64,
2505    total_size: Option<u64>,
2506) {
2507    let Some(total_size) = total_size else {
2508        return;
2509    };
2510
2511    if total_size < DOWNLOAD_PROGRESS_THRESHOLD {
2512        return;
2513    }
2514
2515    let progress_bar =
2516        progress.get_or_insert_with(|| ProgressBar::new(output_config.clone(), total_size));
2517    progress_bar.set_position(bytes_downloaded);
2518}
2519
2520fn print_upload_success(
2521    formatter: &Formatter,
2522    info: &rc_core::ObjectInfo,
2523    src_display: &str,
2524    dst_display: &str,
2525) {
2526    if formatter.is_json() {
2527        print_copy_json(formatter, info, src_display, dst_display);
2528    } else {
2529        let styled_src = formatter.style_file(src_display);
2530        let styled_dst = formatter.style_file(dst_display);
2531        let styled_size = formatter.style_size(&info.size_human.clone().unwrap_or_default());
2532        formatter.println(&format!("{styled_src} -> {styled_dst} ({styled_size})"));
2533    }
2534}
2535
2536async fn upload_file(
2537    client: &S3Client,
2538    src: &Path,
2539    dst: &RemotePath,
2540    args: &CpArgs,
2541    formatter: &Formatter,
2542    encryption: Option<&ObjectEncryptionRequest>,
2543) -> ExitCode {
2544    // Determine destination key
2545    let dst_key = if dst.key.is_empty() || dst.key.ends_with('/') {
2546        // If destination is a directory, use source filename
2547        let filename = src.file_name().unwrap_or_default().to_string_lossy();
2548        format!("{}{}", dst.key, filename)
2549    } else {
2550        dst.key.clone()
2551    };
2552
2553    let target = RemotePath::new(&dst.alias, &dst.bucket, &dst_key);
2554    let src_display = src.display().to_string();
2555    let dst_display = format!("{}/{}/{}", dst.alias, dst.bucket, dst_key);
2556
2557    // Get file size for progress bar decision
2558    let file_size = match std::fs::metadata(src) {
2559        Ok(m) => m.len(),
2560        Err(e) => {
2561            return formatter.fail(
2562                ExitCode::GeneralError,
2563                &format!("Failed to read {src_display}: {e}"),
2564            );
2565        }
2566    };
2567    if args.storage_class.is_some() && file_size > MULTIPART_THRESHOLD {
2568        return formatter.fail(
2569            ExitCode::UnsupportedFeature,
2570            "RustFS beta.10 does not persist storage class for multipart uploads",
2571        );
2572    }
2573    if args.dry_run {
2574        let styled_src = formatter.style_file(&src_display);
2575        let styled_dst = formatter.style_file(&dst_display);
2576        formatter.println(&format!(
2577            "Would copy: {styled_src} -> {styled_dst}{}",
2578            transfer_policy_suffix(args)
2579        ));
2580        return ExitCode::Success;
2581    }
2582
2583    // Determine content type
2584    let guessed_type: Option<String> = mime_guess::from_path(src)
2585        .first()
2586        .map(|m| m.essence_str().to_string());
2587    let content_type = select_upload_content_type(
2588        args.content_type.as_deref(),
2589        guessed_type.as_deref(),
2590        file_size,
2591    );
2592
2593    // Show progress bar for large files
2594    let progress = if file_size > MULTIPART_THRESHOLD {
2595        tracing::debug!(
2596            file_size,
2597            threshold = MULTIPART_THRESHOLD,
2598            "Using multipart upload for large file"
2599        );
2600        Some(ProgressBar::new(formatter.output_config(), file_size))
2601    } else {
2602        tracing::debug!(file_size, "Using single put_object for small file");
2603        None
2604    };
2605
2606    // Upload
2607    let upload_result = match object_write_options(
2608        &args.fidelity,
2609        content_type,
2610        encryption,
2611        args.destination_customer_key.as_ref(),
2612        args.storage_class.clone(),
2613    ) {
2614        Ok(options) => {
2615            client
2616                .put_object_from_path_with_options(&target, src, &options, |bytes_sent| {
2617                    if let Some(ref pb) = progress {
2618                        pb.set_position(bytes_sent);
2619                    }
2620                })
2621                .await
2622        }
2623        Err(error) => Err(error),
2624    };
2625    match upload_result {
2626        Ok(info) => {
2627            if let Some(ref pb) = progress {
2628                pb.finish_and_clear();
2629            }
2630            print_upload_success(formatter, &info, &src_display, &dst_display);
2631            ExitCode::Success
2632        }
2633        Err(e) => {
2634            if let Some(ref pb) = progress {
2635                pb.finish_and_clear();
2636            }
2637            formatter.fail(
2638                exit_code_for_core_error(&e),
2639                &format!("Failed to upload {src_display}: {e}"),
2640            )
2641        }
2642    }
2643}
2644
2645fn select_upload_content_type<'a>(
2646    explicit_type: Option<&'a str>,
2647    guessed_type: Option<&'a str>,
2648    file_size: u64,
2649) -> Option<&'a str> {
2650    if file_size > MULTIPART_THRESHOLD {
2651        explicit_type
2652    } else {
2653        explicit_type.or(guessed_type)
2654    }
2655}
2656
2657async fn upload_directory(
2658    client: &S3Client,
2659    src: &Path,
2660    dst: &RemotePath,
2661    args: &CpArgs,
2662    formatter: &Formatter,
2663    encryption: Option<&ObjectEncryptionRequest>,
2664) -> ExitCode {
2665    use std::fs;
2666
2667    let mut success_count = 0;
2668    let mut error_count = 0;
2669
2670    // Walk directory
2671    fn walk_dir(dir: &Path, base: &Path) -> std::io::Result<Vec<(std::path::PathBuf, String)>> {
2672        let mut files = Vec::new();
2673        for entry in fs::read_dir(dir)? {
2674            let entry = entry?;
2675            let path = entry.path();
2676            if path.is_file() {
2677                let relative = path.strip_prefix(base).unwrap_or(&path);
2678                let relative_str = relative.to_string_lossy().to_string();
2679                files.push((path, relative_str));
2680            } else if path.is_dir() {
2681                files.extend(walk_dir(&path, base)?);
2682            }
2683        }
2684        Ok(files)
2685    }
2686
2687    let files = match walk_dir(src, src) {
2688        Ok(f) => f,
2689        Err(e) => {
2690            return formatter.fail(
2691                ExitCode::GeneralError,
2692                &format!("Failed to read directory: {e}"),
2693            );
2694        }
2695    };
2696
2697    for (file_path, relative_path) in files {
2698        // Build destination key
2699        let dst_key = if dst.key.is_empty() {
2700            relative_path.replace('\\', "/")
2701        } else if dst.key.ends_with('/') {
2702            format!("{}{}", dst.key, relative_path.replace('\\', "/"))
2703        } else {
2704            format!("{}/{}", dst.key, relative_path.replace('\\', "/"))
2705        };
2706
2707        let target = RemotePath::new(&dst.alias, &dst.bucket, &dst_key);
2708
2709        let result = upload_file(client, &file_path, &target, args, formatter, encryption).await;
2710
2711        if result == ExitCode::Success {
2712            success_count += 1;
2713        } else {
2714            error_count += 1;
2715            if !args.continue_on_error {
2716                return result;
2717            }
2718        }
2719    }
2720
2721    if error_count > 0 {
2722        formatter.warning(&format!(
2723            "Completed with errors: {success_count} succeeded, {error_count} failed"
2724        ));
2725        ExitCode::GeneralError
2726    } else {
2727        if !formatter.is_json() {
2728            formatter.success(&format!("Uploaded {success_count} file(s)."));
2729        }
2730        ExitCode::Success
2731    }
2732}
2733
2734async fn copy_s3_to_local(
2735    src: &RemotePath,
2736    dst: &Path,
2737    args: &CpArgs,
2738    formatter: &Formatter,
2739) -> ExitCode {
2740    // Load alias and create client
2741    let alias_manager = match AliasManager::new() {
2742        Ok(am) => am,
2743        Err(e) => {
2744            formatter.error(&format!("Failed to load aliases: {e}"));
2745            return ExitCode::GeneralError;
2746        }
2747    };
2748
2749    let alias = match alias_manager.get(&src.alias) {
2750        Ok(a) => a,
2751        Err(_) => {
2752            return formatter.fail_with_suggestion(
2753                ExitCode::NotFound,
2754                &format!("Alias '{}' not found", src.alias),
2755                "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
2756            );
2757        }
2758    };
2759    let client = match S3Client::new(alias).await {
2760        Ok(c) => c,
2761        Err(e) => {
2762            return formatter.fail(
2763                ExitCode::NetworkError,
2764                &format!("Failed to create S3 client: {e}"),
2765            );
2766        }
2767    };
2768
2769    // Check if source is a prefix (directory-like)
2770    let is_prefix = src.key.is_empty() || src.key.ends_with('/');
2771
2772    if is_prefix || args.recursive {
2773        // Download multiple objects
2774        download_prefix(&client, src, dst, args, formatter).await
2775    } else {
2776        // Download single object
2777        download_file(&client, src, dst, args, formatter).await
2778    }
2779}
2780
2781pub(super) async fn download_file(
2782    client: &S3Client,
2783    src: &RemotePath,
2784    dst: &Path,
2785    args: &CpArgs,
2786    formatter: &Formatter,
2787) -> ExitCode {
2788    let src_display = format!("{}/{}/{}", src.alias, src.bucket, src.key);
2789
2790    // Determine destination path
2791    let dst_path = if dst.is_dir() || dst.to_string_lossy().ends_with('/') {
2792        let filename = src.key.rsplit('/').next().unwrap_or(&src.key);
2793        let filename = match safe_download_relative_path(filename, "", args.local_key_policy()) {
2794            Ok(filename) => filename,
2795            Err(error) => {
2796                return formatter.fail(
2797                    ExitCode::UsageError,
2798                    &format!("Unsafe object key '{}': {error}", src.key),
2799                );
2800            }
2801        };
2802        dst.join(filename)
2803    } else {
2804        dst.to_path_buf()
2805    };
2806
2807    let dst_display = dst_path.display().to_string();
2808
2809    if args.dry_run {
2810        let styled_src = formatter.style_file(&src_display);
2811        let styled_dst = formatter.style_file(&dst_display);
2812        formatter.println(&format!("Would copy: {styled_src} -> {styled_dst}"));
2813        return ExitCode::Success;
2814    }
2815
2816    // Check if destination exists
2817    if dst_path.exists() && !args.overwrite {
2818        return formatter.fail_with_suggestion(
2819            ExitCode::Conflict,
2820            &format!("Destination exists: {dst_display}. Use --overwrite to replace."),
2821            "Retry with --overwrite if replacing the destination file is intended.",
2822        );
2823    }
2824
2825    // Create parent directories
2826    if let Some(parent) = dst_path.parent()
2827        && !parent.exists()
2828        && let Err(e) = std::fs::create_dir_all(parent)
2829    {
2830        return formatter.fail(
2831            ExitCode::GeneralError,
2832            &format!("Failed to create directory: {e}"),
2833        );
2834    }
2835
2836    let output_config = formatter.output_config();
2837    let mut progress = None;
2838
2839    // Download object
2840    let result = client
2841        .download_object_to_path_with_transfer_options(
2842            src,
2843            &dst_path,
2844            &TransferReadOptions {
2845                customer_key: args.source_customer_key.clone(),
2846                ..TransferReadOptions::default()
2847            },
2848            |bytes_downloaded, total_size| {
2849                update_download_progress(
2850                    &mut progress,
2851                    &output_config,
2852                    bytes_downloaded,
2853                    total_size,
2854                );
2855            },
2856        )
2857        .await;
2858
2859    if let Some(ref pb) = progress {
2860        pb.finish_and_clear();
2861    }
2862
2863    match result {
2864        Ok(size) => {
2865            let size = size as i64;
2866
2867            if formatter.is_json() {
2868                let output = CpOutput {
2869                    status: "success",
2870                    source: src_display,
2871                    target: dst_display,
2872                    size_bytes: Some(size),
2873                    size_human: Some(humansize::format_size(size as u64, humansize::BINARY)),
2874                    version_id: None,
2875                    source_version_id: None,
2876                };
2877                formatter.json(&output);
2878            } else {
2879                let styled_src = formatter.style_file(&src_display);
2880                let styled_dst = formatter.style_file(&dst_display);
2881                let styled_size =
2882                    formatter.style_size(&humansize::format_size(size as u64, humansize::BINARY));
2883                formatter.println(&format!("{styled_src} -> {styled_dst} ({styled_size})"));
2884            }
2885            ExitCode::Success
2886        }
2887        Err(e) => {
2888            let err_str = e.to_string();
2889            if err_str.contains("NotFound") || err_str.contains("NoSuchKey") {
2890                formatter.fail_with_suggestion(
2891                    ExitCode::NotFound,
2892                    &format!("Object not found: {src_display}"),
2893                    "Check the object key and bucket path, then retry the copy command.",
2894                )
2895            } else {
2896                formatter.fail(
2897                    ExitCode::NetworkError,
2898                    &format!("Failed to download {src_display}: {e}"),
2899                )
2900            }
2901        }
2902    }
2903}
2904
2905async fn download_prefix(
2906    client: &S3Client,
2907    src: &RemotePath,
2908    dst: &Path,
2909    args: &CpArgs,
2910    formatter: &Formatter,
2911) -> ExitCode {
2912    use rc_core::ListOptions;
2913
2914    let mut success_count = 0;
2915    let mut error_count = 0;
2916    let mut continuation_token: Option<String> = None;
2917
2918    loop {
2919        let options = ListOptions {
2920            recursive: true,
2921            max_keys: Some(1000),
2922            continuation_token: continuation_token.clone(),
2923            ..Default::default()
2924        };
2925
2926        match client.list_objects(src, options).await {
2927            Ok(result) => {
2928                for item in result.items {
2929                    if item.is_dir {
2930                        continue;
2931                    }
2932
2933                    // Calculate relative path from prefix
2934                    let relative_path = match safe_download_relative_path(
2935                        &item.key,
2936                        &src.key,
2937                        args.local_key_policy(),
2938                    ) {
2939                        Ok(path) => path,
2940                        Err(error) => {
2941                            error_count += 1;
2942                            formatter.error(&format!(
2943                                "Refusing unsafe object key '{}': {error}",
2944                                item.key
2945                            ));
2946                            if !args.continue_on_error {
2947                                return ExitCode::UsageError;
2948                            }
2949                            continue;
2950                        }
2951                    };
2952                    let dst_path = match safe_download_destination(dst, &relative_path).await {
2953                        Ok(path) => path,
2954                        Err(error) => {
2955                            error_count += 1;
2956                            formatter.error(&format!(
2957                                "Refusing unsafe destination for '{}': {error}",
2958                                item.key
2959                            ));
2960                            if !args.continue_on_error {
2961                                return ExitCode::UsageError;
2962                            }
2963                            continue;
2964                        }
2965                    };
2966
2967                    let obj_src = RemotePath::new(&src.alias, &src.bucket, &item.key);
2968                    let result = download_file(client, &obj_src, &dst_path, args, formatter).await;
2969
2970                    if result == ExitCode::Success {
2971                        success_count += 1;
2972                    } else {
2973                        error_count += 1;
2974                        if !args.continue_on_error {
2975                            return result;
2976                        }
2977                    }
2978                }
2979
2980                if result.truncated {
2981                    continuation_token = result.continuation_token;
2982                } else {
2983                    break;
2984                }
2985            }
2986            Err(e) => {
2987                return formatter.fail(
2988                    ExitCode::NetworkError,
2989                    &format!("Failed to list objects: {e}"),
2990                );
2991            }
2992        }
2993    }
2994
2995    if error_count > 0 {
2996        formatter.warning(&format!(
2997            "Completed with errors: {success_count} succeeded, {error_count} failed"
2998        ));
2999        ExitCode::GeneralError
3000    } else if success_count == 0 {
3001        formatter.warning("No objects found to download.");
3002        ExitCode::Success
3003    } else {
3004        if !formatter.is_json() {
3005            formatter.success(&format!("Downloaded {success_count} file(s)."));
3006        }
3007        ExitCode::Success
3008    }
3009}
3010
3011pub(super) fn safe_download_relative_path(
3012    key: &str,
3013    prefix: &str,
3014    policy: ObjectKeyPolicy,
3015) -> Result<PathBuf, String> {
3016    relative_local_path_from_key(key, prefix, policy).map_err(|error| error.to_string())
3017}
3018
3019pub(super) async fn safe_download_destination(
3020    root: &Path,
3021    relative: &Path,
3022) -> Result<PathBuf, String> {
3023    let mut destination = root.to_path_buf();
3024    for component in relative.components() {
3025        let std::path::Component::Normal(component) = component else {
3026            return Err("destination path contains a non-normal component".to_string());
3027        };
3028        destination.push(component);
3029        match tokio::fs::symlink_metadata(&destination).await {
3030            Ok(metadata) if metadata.file_type().is_symlink() => {
3031                return Err(format!(
3032                    "destination component '{}' is a symbolic link",
3033                    destination.display()
3034                ));
3035            }
3036            Ok(_) => {}
3037            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
3038            Err(error) => {
3039                return Err(format!(
3040                    "failed to inspect destination '{}': {error}",
3041                    destination.display()
3042                ));
3043            }
3044        }
3045    }
3046
3047    Ok(destination)
3048}
3049
3050async fn copy_s3_to_s3_prepared(
3051    src: &RemotePath,
3052    dst: &RemotePath,
3053    args: &CpArgs,
3054    formatter: &Formatter,
3055    encryption: Option<&ObjectEncryptionRequest>,
3056) -> ExitCode {
3057    if args.source_customer_key.is_some() || args.destination_customer_key.is_some() {
3058        return formatter.fail(
3059            ExitCode::UnsupportedFeature,
3060            "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467",
3061        );
3062    }
3063
3064    let alias_manager = match AliasManager::new() {
3065        Ok(am) => am,
3066        Err(e) => {
3067            formatter.error(&format!("Failed to load aliases: {e}"));
3068            return ExitCode::GeneralError;
3069        }
3070    };
3071
3072    // Same-alias copies use server-side CopyObject. Different aliases download
3073    // through a temporary file and upload with the destination credentials.
3074    let source_alias = match alias_manager.get(&src.alias) {
3075        Ok(a) => a,
3076        Err(_) => {
3077            return formatter.fail_with_suggestion(
3078                ExitCode::NotFound,
3079                &format!("Alias '{}' not found", src.alias),
3080                "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
3081            );
3082        }
3083    };
3084    let source_client = match S3Client::new(source_alias).await {
3085        Ok(c) => c,
3086        Err(e) => {
3087            return formatter.fail(
3088                ExitCode::NetworkError,
3089                &format!("Failed to create S3 client: {e}"),
3090            );
3091        }
3092    };
3093    let target_client;
3094    let target_client_ref = if src.alias == dst.alias {
3095        &source_client
3096    } else {
3097        let destination_alias = match alias_manager.get(&dst.alias) {
3098            Ok(a) => a,
3099            Err(_) => {
3100                return formatter.fail_with_suggestion(
3101                    ExitCode::NotFound,
3102                    &format!("Alias '{}' not found", dst.alias),
3103                    "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
3104                );
3105            }
3106        };
3107        target_client = match S3Client::new(destination_alias).await {
3108            Ok(c) => c,
3109            Err(e) => {
3110                return formatter.fail(
3111                    ExitCode::NetworkError,
3112                    &format!("Failed to create destination S3 client: {e}"),
3113                );
3114            }
3115        };
3116        &target_client
3117    };
3118
3119    let src_display = format!("{}/{}/{}", src.alias, src.bucket, src.key);
3120    let dst_display = format!("{}/{}/{}", dst.alias, dst.bucket, dst.key);
3121
3122    if args.dry_run && args.storage_class.is_none() {
3123        let styled_src = formatter.style_file(&src_display);
3124        let styled_dst = formatter.style_file(&dst_display);
3125        formatter.println(&format!(
3126            "Would copy: {styled_src} -> {styled_dst}{}",
3127            transfer_policy_suffix(args)
3128        ));
3129        return ExitCode::Success;
3130    }
3131
3132    let source_info = match source_client.head_object(src).await {
3133        Ok(info) => info,
3134        Err(Error::NotFound(_)) => {
3135            return formatter.fail_with_suggestion(
3136                ExitCode::NotFound,
3137                &format!("Source not found: {src_display}"),
3138                "Check the source bucket and object key, then retry the copy command.",
3139            );
3140        }
3141        Err(error) => {
3142            return formatter.fail(
3143                exit_code_for_core_error(&error),
3144                &format!("Failed to inspect source object: {error}"),
3145            );
3146        }
3147    };
3148    let source_size = source_info
3149        .size_bytes
3150        .and_then(|size| u64::try_from(size).ok());
3151    if args.storage_class.is_some()
3152        && source_size.is_none_or(|size| {
3153            if src.alias == dst.alias {
3154                rc_core::requires_multipart_copy(size)
3155            } else {
3156                size > MULTIPART_THRESHOLD
3157            }
3158        })
3159    {
3160        return formatter.fail(
3161            ExitCode::UnsupportedFeature,
3162            "RustFS beta.10 does not persist storage class for multipart or unknown-size copies",
3163        );
3164    }
3165    if args.dry_run {
3166        let styled_src = formatter.style_file(&src_display);
3167        let styled_dst = formatter.style_file(&dst_display);
3168        formatter.println(&format!(
3169            "Would copy: {styled_src} -> {styled_dst}{}",
3170            transfer_policy_suffix(args)
3171        ));
3172        return ExitCode::Success;
3173    }
3174
3175    let cancellation = MultipartCopyCancellation::new();
3176    let transfer_cancellation = TransferCancellation::new();
3177    let signal_task = tokio::spawn({
3178        let cancellation = cancellation.clone();
3179        let transfer_cancellation = transfer_cancellation.clone();
3180        async move {
3181            if tokio::signal::ctrl_c().await.is_ok() {
3182                cancellation.cancel();
3183                transfer_cancellation.cancel();
3184            }
3185        }
3186    });
3187    let ignore_progress = |_: u64| {};
3188    let copy = perform_planned_remote_copy(
3189        &source_client,
3190        target_client_ref,
3191        src,
3192        dst,
3193        &source_info,
3194        encryption,
3195        &cancellation,
3196        &ignore_progress,
3197        args,
3198    );
3199    tokio::pin!(copy);
3200    let result = tokio::select! {
3201        biased;
3202        _ = transfer_cancellation.cancelled() => {
3203            // Do not drop an in-flight request. Multipart copy observes its own token and performs
3204            // cleanup; CopyObject is allowed to settle before the command reports interruption.
3205            let _ = copy.await;
3206            Err(Error::Interrupted("Copy interrupted".to_string()))
3207        }
3208        result = &mut copy => {
3209            if transfer_cancellation.is_cancelled() {
3210                Err(Error::Interrupted("Copy interrupted".to_string()))
3211            } else {
3212                result
3213            }
3214        }
3215    };
3216    signal_task.abort();
3217    let _ = signal_task.await;
3218
3219    match result {
3220        Ok(result) => {
3221            if formatter.is_json() {
3222                print_copy_json(formatter, &result.object, &src_display, &dst_display);
3223            } else {
3224                let styled_src = formatter.style_file(&src_display);
3225                let styled_dst = formatter.style_file(&dst_display);
3226                let styled_size =
3227                    formatter.style_size(&result.object.size_human.unwrap_or_else(|| {
3228                        humansize::format_size(result.bytes_copied, humansize::BINARY)
3229                    }));
3230                let upload = result
3231                    .upload_id
3232                    .map(|upload_id| format!(" upload-id={upload_id}"))
3233                    .unwrap_or_default();
3234                formatter.println(&format!(
3235                    "{styled_src} -> {styled_dst} ({styled_size}){upload}"
3236                ));
3237            }
3238            ExitCode::Success
3239        }
3240        Err(error) => {
3241            if matches!(error, Error::NotFound(_) | Error::VersionNotFound { .. }) {
3242                formatter.fail_with_suggestion(
3243                    ExitCode::NotFound,
3244                    &format!("Source not found: {src_display}"),
3245                    "Check the source bucket and object key, then retry the copy command.",
3246                )
3247            } else {
3248                formatter.fail(
3249                    exit_code_for_core_error(&error),
3250                    &format!("Failed to copy: {error}"),
3251                )
3252            }
3253        }
3254    }
3255}
3256
3257fn print_copy_json(formatter: &Formatter, info: &rc_core::ObjectInfo, source: &str, target: &str) {
3258    if info.version_id.is_some() || info.source_version_id.is_some() {
3259        formatter.json(&V3SuccessEnvelope::versioned_objects(VersionCopyData {
3260            operation: "copy",
3261            source: source.to_string(),
3262            target: target.to_string(),
3263            source_version_id: info.source_version_id.clone(),
3264            version_id: info.version_id.clone(),
3265            size_bytes: info.size_bytes,
3266            size_human: info.size_human.clone(),
3267        }));
3268    } else {
3269        formatter.json(&CpOutput {
3270            status: "success",
3271            source: source.to_string(),
3272            target: target.to_string(),
3273            size_bytes: info.size_bytes,
3274            size_human: info.size_human.clone(),
3275            version_id: None,
3276            source_version_id: None,
3277        });
3278    }
3279}
3280
3281fn parse_kms_target(value: &str) -> Result<(String, String), String> {
3282    let (target, key_id) = value
3283        .split_once('=')
3284        .ok_or_else(|| "Expected TARGET=KMS_KEY_ID for --enc-kms".to_string())?;
3285
3286    if target.is_empty() || key_id.is_empty() {
3287        return Err("Expected TARGET=KMS_KEY_ID for --enc-kms".to_string());
3288    }
3289
3290    Ok((target.to_string(), key_id.to_string()))
3291}
3292
3293pub(crate) fn parse_destination_encryption(
3294    enc_s3: &[String],
3295    enc_kms: &[String],
3296    target: &ParsedPath,
3297) -> Result<Option<ObjectEncryptionRequest>, String> {
3298    if enc_s3.is_empty() && enc_kms.is_empty() {
3299        return Ok(None);
3300    }
3301
3302    let remote = match target {
3303        ParsedPath::Remote(remote) => remote,
3304        ParsedPath::Local(_) => {
3305            return Err("Destination encryption flags must reference a remote destination".into());
3306        }
3307    };
3308
3309    let target_display = remote.to_string();
3310    let s3_matches = enc_s3.iter().any(|value| value == &target_display);
3311    let kms_targets = enc_kms
3312        .iter()
3313        .map(|value| parse_kms_target(value))
3314        .collect::<Result<Vec<_>, _>>()?;
3315    let kms_match = kms_targets
3316        .iter()
3317        .find(|(candidate, _)| candidate == &target_display);
3318
3319    if !enc_s3.is_empty() && !s3_matches {
3320        return Err(format!(
3321            "--enc-s3 target must exactly match the remote destination: {target_display}"
3322        ));
3323    }
3324
3325    if !enc_kms.is_empty() && kms_match.is_none() {
3326        return Err(format!(
3327            "--enc-kms target must exactly match the remote destination: {target_display}"
3328        ));
3329    }
3330
3331    match (s3_matches, kms_match) {
3332        (true, Some(_)) => Err(format!(
3333            "--enc-s3 and --enc-kms cannot target the same destination: {target_display}"
3334        )),
3335        (true, None) => Ok(Some(ObjectEncryptionRequest::SseS3)),
3336        (false, Some((_, key_id))) => Ok(Some(ObjectEncryptionRequest::SseKms {
3337            key_id: key_id.clone(),
3338        })),
3339        (false, None) => Ok(None),
3340    }
3341}
3342
3343#[cfg(test)]
3344mod tests {
3345    use super::*;
3346    use rc_core::{Alias, ConfigManager};
3347    use tempfile::TempDir;
3348
3349    fn temp_alias_manager() -> (AliasManager, TempDir) {
3350        let temp_dir = TempDir::new().expect("create temp dir");
3351        let config_path = temp_dir.path().join("config.toml");
3352        let config_manager = ConfigManager::with_path(config_path);
3353        let alias_manager = AliasManager::with_config_manager(config_manager);
3354        (alias_manager, temp_dir)
3355    }
3356
3357    #[test]
3358    fn test_parse_local_path() {
3359        let result = parse_path("./file.txt").unwrap();
3360        assert!(matches!(result, ParsedPath::Local(_)));
3361    }
3362
3363    #[test]
3364    fn test_parse_remote_path() {
3365        let result = parse_path("myalias/bucket/file.txt").unwrap();
3366        assert!(matches!(result, ParsedPath::Remote(_)));
3367    }
3368
3369    #[test]
3370    fn test_parse_local_absolute_path() {
3371        // Use platform-appropriate absolute path
3372        #[cfg(unix)]
3373        let path = "/home/user/file.txt";
3374        #[cfg(windows)]
3375        let path = "C:\\Users\\user\\file.txt";
3376
3377        let result = parse_path(path).unwrap();
3378        assert!(matches!(result, ParsedPath::Local(_)));
3379        if let ParsedPath::Local(p) = result {
3380            assert!(p.is_absolute());
3381        }
3382    }
3383
3384    #[test]
3385    fn test_parse_local_relative_path() {
3386        let result = parse_path("../file.txt").unwrap();
3387        assert!(matches!(result, ParsedPath::Local(_)));
3388    }
3389
3390    #[test]
3391    fn test_parse_remote_path_bucket_only() {
3392        let result = parse_path("myalias/bucket/").unwrap();
3393        assert!(matches!(result, ParsedPath::Remote(_)));
3394        if let ParsedPath::Remote(r) = result {
3395            assert_eq!(r.alias, "myalias");
3396            assert_eq!(r.bucket, "bucket");
3397            assert!(r.key.is_empty());
3398        }
3399    }
3400
3401    #[test]
3402    fn test_parse_remote_path_with_deep_key() {
3403        let result = parse_path("myalias/bucket/dir1/dir2/file.txt").unwrap();
3404        assert!(matches!(result, ParsedPath::Remote(_)));
3405        if let ParsedPath::Remote(r) = result {
3406            assert_eq!(r.alias, "myalias");
3407            assert_eq!(r.bucket, "bucket");
3408            assert_eq!(r.key, "dir1/dir2/file.txt");
3409        }
3410    }
3411
3412    #[test]
3413    fn test_download_progress_created_for_large_transfer() {
3414        let output_config = OutputConfig::default();
3415        let mut progress = None;
3416
3417        update_download_progress(
3418            &mut progress,
3419            &output_config,
3420            1024,
3421            Some(DOWNLOAD_PROGRESS_THRESHOLD),
3422        );
3423
3424        let progress = progress.expect("large download should create progress bar");
3425        assert!(progress.is_visible());
3426        progress.finish_and_clear();
3427    }
3428
3429    #[test]
3430    fn test_download_progress_skips_small_transfer() {
3431        let output_config = OutputConfig::default();
3432        let mut progress = None;
3433
3434        update_download_progress(
3435            &mut progress,
3436            &output_config,
3437            1024,
3438            Some(DOWNLOAD_PROGRESS_THRESHOLD - 1),
3439        );
3440
3441        assert!(progress.is_none());
3442    }
3443
3444    #[test]
3445    fn test_download_progress_skips_unknown_total_size() {
3446        let output_config = OutputConfig::default();
3447        let mut progress = None;
3448
3449        update_download_progress(&mut progress, &output_config, 1024, None);
3450
3451        assert!(progress.is_none());
3452    }
3453
3454    #[test]
3455    fn test_download_progress_respects_no_progress_config() {
3456        let output_config = OutputConfig {
3457            no_progress: true,
3458            ..Default::default()
3459        };
3460        let mut progress = None;
3461
3462        update_download_progress(
3463            &mut progress,
3464            &output_config,
3465            1024,
3466            Some(DOWNLOAD_PROGRESS_THRESHOLD),
3467        );
3468
3469        let progress = progress.expect("large download should create progress state");
3470        assert!(!progress.is_visible());
3471    }
3472
3473    #[test]
3474    fn download_relative_path_preserves_safe_nested_keys() {
3475        let relative = safe_download_relative_path(
3476            "reports/2026/july/data.csv",
3477            "reports/",
3478            ObjectKeyPolicy::Logical,
3479        )
3480        .expect("safe key should resolve");
3481
3482        assert_eq!(
3483            relative,
3484            PathBuf::from("2026").join("july").join("data.csv")
3485        );
3486    }
3487
3488    #[test]
3489    fn download_relative_path_rejects_traversal_and_absolute_keys() {
3490        for key in [
3491            "reports/../../escaped",
3492            "reports/..\\..\\escaped",
3493            "/absolute/path",
3494        ] {
3495            assert!(
3496                safe_download_relative_path(key, "reports/", ObjectKeyPolicy::Logical).is_err(),
3497                "unsafe key should be rejected: {key}"
3498            );
3499        }
3500    }
3501
3502    #[test]
3503    fn download_relative_path_accepts_colon_keys_on_logical_destinations() {
3504        let relative = safe_download_relative_path(
3505            "loki/fake/deadbeef/19f6abd9af4:19f6abe0e77:499628ff",
3506            "loki/",
3507            ObjectKeyPolicy::Logical,
3508        )
3509        .expect("colon keys are valid Unix file names");
3510
3511        assert_eq!(
3512            relative,
3513            PathBuf::from("fake")
3514                .join("deadbeef")
3515                .join("19f6abd9af4:19f6abe0e77:499628ff")
3516        );
3517    }
3518
3519    #[test]
3520    fn download_relative_path_rejects_colon_keys_when_portable_names_requested() {
3521        for key in [
3522            "reports/C:/escaped",
3523            "reports/safe:stream",
3524            "reports/CON.txt",
3525            "reports/trailing.",
3526        ] {
3527            assert!(
3528                safe_download_relative_path(key, "reports/", ObjectKeyPolicy::WindowsPortable)
3529                    .is_err(),
3530                "portable-unsafe key should be rejected: {key}"
3531            );
3532        }
3533    }
3534
3535    #[cfg(unix)]
3536    #[tokio::test]
3537    async fn download_destination_rejects_existing_symlink_components() {
3538        use std::os::unix::fs::symlink;
3539
3540        let root = tempfile::tempdir().expect("create destination root");
3541        let outside = tempfile::tempdir().expect("create outside directory");
3542        symlink(outside.path(), root.path().join("linked")).expect("create test symlink");
3543
3544        let result =
3545            safe_download_destination(root.path(), &PathBuf::from("linked/file.txt")).await;
3546
3547        assert!(result.is_err());
3548    }
3549
3550    #[cfg(unix)]
3551    #[tokio::test]
3552    async fn download_destination_rejects_existing_symlink_file() {
3553        use std::os::unix::fs::symlink;
3554
3555        let root = tempfile::tempdir().expect("create destination root");
3556        let outside = tempfile::tempdir().expect("create outside directory");
3557        let outside_file = outside.path().join("file.txt");
3558        std::fs::write(&outside_file, b"outside").expect("write outside file");
3559        symlink(&outside_file, root.path().join("file.txt")).expect("create test symlink");
3560
3561        let result = safe_download_destination(root.path(), &PathBuf::from("file.txt")).await;
3562
3563        assert!(result.is_err());
3564    }
3565
3566    #[cfg(unix)]
3567    #[test]
3568    fn recursive_source_collection_rejects_symbolic_links() {
3569        use std::os::unix::fs::symlink;
3570
3571        let root = tempfile::tempdir().expect("create source root");
3572        let outside = tempfile::NamedTempFile::new().expect("create outside file");
3573        symlink(outside.path(), root.path().join("linked.txt")).expect("create source symlink");
3574        let mut files = Vec::new();
3575
3576        let result = collect_local_files(root.path(), root.path(), &mut files);
3577
3578        assert!(result.is_err());
3579        assert!(files.is_empty());
3580    }
3581
3582    #[test]
3583    fn test_select_upload_content_type_uses_guess_for_small_files() {
3584        let selected =
3585            select_upload_content_type(None, Some("text/plain"), MULTIPART_THRESHOLD - 1);
3586
3587        assert_eq!(selected, Some("text/plain"));
3588    }
3589
3590    #[test]
3591    fn test_select_upload_content_type_skips_guess_for_multipart_files() {
3592        let selected =
3593            select_upload_content_type(None, Some("text/plain"), MULTIPART_THRESHOLD + 1);
3594
3595        assert_eq!(selected, None);
3596    }
3597
3598    #[test]
3599    fn test_select_upload_content_type_uses_guess_at_multipart_boundary() {
3600        let selected = select_upload_content_type(None, Some("text/plain"), MULTIPART_THRESHOLD);
3601
3602        assert_eq!(selected, Some("text/plain"));
3603    }
3604
3605    #[test]
3606    fn test_select_upload_content_type_keeps_explicit_type_for_multipart_files() {
3607        let selected = select_upload_content_type(
3608            Some("application/octet-stream"),
3609            Some("text/plain"),
3610            MULTIPART_THRESHOLD + 1,
3611        );
3612
3613        assert_eq!(selected, Some("application/octet-stream"));
3614    }
3615
3616    #[test]
3617    fn test_parse_cp_path_prefers_existing_local_path_when_alias_missing() {
3618        let (alias_manager, temp_dir) = temp_alias_manager();
3619        let full = temp_dir.path().join("issue-2094-local").join("file.txt");
3620        let full_str = full.to_string_lossy().to_string();
3621
3622        if let Some(parent) = full.parent() {
3623            std::fs::create_dir_all(parent).expect("create parent dirs");
3624        }
3625        std::fs::write(&full, b"test").expect("write local file");
3626
3627        let parsed = parse_cp_path(&full_str, Some(&alias_manager)).expect("parse path");
3628        assert!(matches!(parsed, ParsedPath::Local(_)));
3629    }
3630
3631    #[test]
3632    fn test_parse_cp_path_keeps_remote_when_alias_exists() {
3633        let (alias_manager, _temp_dir) = temp_alias_manager();
3634        alias_manager
3635            .set(Alias::new("target", "http://localhost:9000", "a", "b"))
3636            .expect("set alias");
3637
3638        let parsed = parse_cp_path("target/bucket/file.txt", Some(&alias_manager))
3639            .expect("parse remote path");
3640        assert!(matches!(parsed, ParsedPath::Remote(_)));
3641    }
3642
3643    #[test]
3644    fn test_parse_cp_path_keeps_remote_when_local_missing() {
3645        let (alias_manager, _temp_dir) = temp_alias_manager();
3646        let parsed = parse_cp_path("missing/bucket/file.txt", Some(&alias_manager))
3647            .expect("parse remote path");
3648        assert!(matches!(parsed, ParsedPath::Remote(_)));
3649    }
3650
3651    #[test]
3652    fn test_cp_args_defaults() {
3653        let args = CpArgs::single("src", "dst");
3654        assert!(args.overwrite);
3655        assert!(!args.recursive);
3656        assert!(!args.dry_run);
3657        assert_eq!(args.concurrency, None);
3658        assert_eq!(args.retry_attempts, None);
3659        let controls = build_transfer_controls(&args).expect("default controls");
3660        assert_eq!(controls.concurrency, DEFAULT_TRANSFER_CONCURRENCY);
3661        assert_eq!(controls.retry.max_attempts, DEFAULT_RETRY_ATTEMPTS);
3662        assert_eq!(
3663            controls.retry.initial_backoff_ms,
3664            DEFAULT_RETRY_INITIAL_BACKOFF_MS
3665        );
3666        assert_eq!(controls.retry.max_backoff_ms, DEFAULT_RETRY_MAX_BACKOFF_MS);
3667    }
3668
3669    #[test]
3670    fn explicit_default_transfer_controls_enable_planner_routing() {
3671        let defaults = CpArgs::single("src", "dst");
3672        assert!(!uses_transfer_planner(&defaults));
3673
3674        let mut explicit_concurrency = defaults.clone();
3675        explicit_concurrency.concurrency = Some(DEFAULT_TRANSFER_CONCURRENCY);
3676        let mut explicit_attempts = defaults.clone();
3677        explicit_attempts.retry_attempts = Some(DEFAULT_RETRY_ATTEMPTS);
3678        let mut explicit_initial_backoff = defaults.clone();
3679        explicit_initial_backoff.retry_initial_backoff_ms = Some(DEFAULT_RETRY_INITIAL_BACKOFF_MS);
3680        let mut explicit_max_backoff = defaults;
3681        explicit_max_backoff.retry_max_backoff_ms = Some(DEFAULT_RETRY_MAX_BACKOFF_MS);
3682
3683        for args in [
3684            explicit_concurrency,
3685            explicit_attempts,
3686            explicit_initial_backoff,
3687            explicit_max_backoff,
3688        ] {
3689            assert!(uses_transfer_planner(&args));
3690        }
3691    }
3692
3693    #[test]
3694    fn planned_client_aliases_are_deduplicated() {
3695        let first = TransferCandidate {
3696            payload: CpOperation::LocalToRemote {
3697                source: PathBuf::from("first.txt"),
3698                target: RemotePath::new("shared", "bucket", "first.txt"),
3699                encryption: None,
3700            },
3701            source: "first.txt".to_string(),
3702            target: "shared/bucket/first.txt".to_string(),
3703            relative_path: "first.txt".to_string(),
3704            modified: None,
3705            size_bytes: Some(1),
3706        };
3707        let second = TransferCandidate {
3708            payload: CpOperation::RemoteToLocal {
3709                source: RemotePath::new("shared", "bucket", "second.txt"),
3710                target: PathBuf::from("second.txt"),
3711            },
3712            source: "shared/bucket/second.txt".to_string(),
3713            target: "second.txt".to_string(),
3714            relative_path: "second.txt".to_string(),
3715            modified: None,
3716            size_bytes: Some(2),
3717        };
3718
3719        assert_eq!(
3720            planned_client_aliases(&[first, second])
3721                .into_iter()
3722                .collect::<Vec<_>>(),
3723            ["shared"]
3724        );
3725    }
3726
3727    #[test]
3728    fn planned_client_aliases_include_both_sides_of_a_cross_alias_copy() {
3729        let candidate = TransferCandidate {
3730            payload: CpOperation::RemoteToRemote {
3731                source: RemotePath::new("alpha", "source", "file.txt"),
3732                target: RemotePath::new("beta", "target", "file.txt"),
3733                source_info: Box::new(ObjectInfo::file("file.txt", 4)),
3734                encryption: None,
3735            },
3736            source: "alpha/source/file.txt".to_string(),
3737            target: "beta/target/file.txt".to_string(),
3738            relative_path: "file.txt".to_string(),
3739            modified: None,
3740            size_bytes: Some(4),
3741        };
3742
3743        assert_eq!(
3744            planned_client_aliases(&[candidate])
3745                .into_iter()
3746                .collect::<Vec<_>>(),
3747            ["alpha", "beta"]
3748        );
3749    }
3750
3751    #[test]
3752    fn planned_client_aliases_keep_same_alias_remote_copy_on_one_alias() {
3753        let candidate = TransferCandidate {
3754            payload: CpOperation::RemoteToRemote {
3755                source: RemotePath::new("shared", "source", "file.txt"),
3756                target: RemotePath::new("shared", "target", "file.txt"),
3757                source_info: Box::new(ObjectInfo::file("file.txt", 4)),
3758                encryption: None,
3759            },
3760            source: "shared/source/file.txt".to_string(),
3761            target: "shared/target/file.txt".to_string(),
3762            relative_path: "file.txt".to_string(),
3763            modified: None,
3764            size_bytes: Some(4),
3765        };
3766
3767        assert_eq!(
3768            planned_client_aliases(&[candidate])
3769                .into_iter()
3770                .collect::<Vec<_>>(),
3771            ["shared"]
3772        );
3773    }
3774
3775    #[tokio::test]
3776    async fn planning_client_reuses_one_connection_pool_per_alias() {
3777        let (alias_manager, _temp_dir) = temp_alias_manager();
3778        alias_manager
3779            .set(Alias::new(
3780                "shared",
3781                "http://localhost:9000",
3782                "access",
3783                "secret",
3784            ))
3785            .expect("set alias");
3786        let mut clients = HashMap::new();
3787
3788        let first = planning_client(&mut clients, &alias_manager, "shared")
3789            .await
3790            .expect("create first client");
3791        let second = planning_client(&mut clients, &alias_manager, "shared")
3792            .await
3793            .expect("reuse first client");
3794
3795        assert_eq!(clients.len(), 1);
3796        assert!(Arc::ptr_eq(&first, &second));
3797    }
3798
3799    #[test]
3800    fn planned_remote_copy_uses_candidate_size_for_multipart_guard() {
3801        let candidate = TransferCandidate {
3802            payload: CpOperation::RemoteToRemote {
3803                source: RemotePath::new("shared", "source", "large.bin"),
3804                target: RemotePath::new("shared", "target", "large.bin"),
3805                source_info: Box::new(ObjectInfo::file(
3806                    "large.bin",
3807                    (MAX_SINGLE_COPY_SIZE + 1) as i64,
3808                )),
3809                encryption: None,
3810            },
3811            source: "shared/source/large.bin".to_string(),
3812            target: "shared/target/large.bin".to_string(),
3813            relative_path: "large.bin".to_string(),
3814            modified: None,
3815            size_bytes: Some(MAX_SINGLE_COPY_SIZE + 1),
3816        };
3817
3818        assert!(requires_multipart_copy(candidate.size_bytes));
3819    }
3820
3821    #[test]
3822    fn planned_remote_copy_uses_exact_five_gib_boundary() {
3823        assert!(!requires_multipart_copy(Some(MAX_SINGLE_COPY_SIZE)));
3824        assert!(requires_multipart_copy(Some(MAX_SINGLE_COPY_SIZE + 1)));
3825    }
3826
3827    #[test]
3828    fn planned_progress_replaces_attempt_bytes_instead_of_double_counting() {
3829        let progress = PlannedCopyProgress::new(
3830            OutputConfig {
3831                no_progress: true,
3832                ..OutputConfig::default()
3833            },
3834            20,
3835        );
3836        let first = ("source/a".to_string(), "target/a".to_string());
3837        let second = ("source/b".to_string(), "target/b".to_string());
3838
3839        progress.set(&first, 7);
3840        progress.set(&second, 5);
3841        progress.reset(&first);
3842        progress.set(&first, 15);
3843
3844        let positions = progress
3845            .positions
3846            .lock()
3847            .expect("planned copy progress lock");
3848        assert_eq!(positions.get(&first), Some(&15));
3849        assert_eq!(positions.get(&second), Some(&5));
3850        assert_eq!(
3851            positions.values().copied().fold(0_u64, u64::saturating_add),
3852            20
3853        );
3854    }
3855
3856    #[test]
3857    fn recursive_remote_mapping_preserves_source_relative_keys() {
3858        let source = RemotePath::new("shared", "source", "src/");
3859        let target = RemotePath::new("shared", "destination", "archive/");
3860
3861        let (destination, relative) = recursive_remote_target(
3862            &source,
3863            &target,
3864            "src/nested/report.csv",
3865            false,
3866            ObjectKeyPolicy::for_remote_destination(),
3867        )
3868        .expect("map recursive object");
3869
3870        assert_eq!(relative, "nested/report.csv");
3871        assert_eq!(destination.key, "archive/nested/report.csv");
3872    }
3873
3874    #[test]
3875    fn recursive_bucket_root_mapping_keeps_the_full_object_key() {
3876        let source = RemotePath::new("shared", "source", "");
3877        let target = RemotePath::new("shared", "destination", "archive/");
3878
3879        let (destination, relative) = recursive_remote_target(
3880            &source,
3881            &target,
3882            "nested/report.csv",
3883            false,
3884            ObjectKeyPolicy::for_remote_destination(),
3885        )
3886        .expect("map bucket object");
3887
3888        assert_eq!(relative, "nested/report.csv");
3889        assert_eq!(destination.key, "archive/nested/report.csv");
3890    }
3891
3892    #[test]
3893    fn recursive_remote_mapping_rejects_unsafe_listed_keys() {
3894        let source = RemotePath::new("shared", "source", "src/");
3895        let target = RemotePath::new("shared", "destination", "archive/");
3896
3897        for object_key in [
3898            "/absolute.txt",
3899            "src/../escape.txt",
3900            "src\\escape.txt",
3901            "src/control\u{0007}.txt",
3902        ] {
3903            assert!(
3904                recursive_remote_target(
3905                    &source,
3906                    &target,
3907                    object_key,
3908                    false,
3909                    ObjectKeyPolicy::for_remote_destination(),
3910                )
3911                .is_err(),
3912                "unsafe listed key should be rejected: {object_key:?}"
3913            );
3914        }
3915    }
3916
3917    #[test]
3918    fn remote_child_rejects_unsafe_relative_keys() {
3919        let target = RemotePath::new("shared", "destination", "archive/");
3920
3921        for relative in [
3922            "../escape.txt",
3923            "nested\\escape.txt",
3924            "nested/control\u{0007}.txt",
3925        ] {
3926            assert!(
3927                remote_child(&target, relative, ObjectKeyPolicy::for_remote_destination(),)
3928                    .is_err(),
3929                "unsafe relative key should be rejected: {relative:?}"
3930            );
3931        }
3932    }
3933
3934    #[test]
3935    fn remote_target_rejects_absolute_prefixes() {
3936        let target = RemotePath::new("shared", "destination", "/archive/");
3937
3938        assert!(
3939            normalize_remote_target(&target, ObjectKeyPolicy::for_remote_destination()).is_err()
3940        );
3941    }
3942
3943    #[test]
3944    fn recursive_remote_overlap_is_boundary_aware_and_symmetric() {
3945        let source = RemotePath::new("shared", "bucket", "src/");
3946        let child = RemotePath::new("shared", "bucket", "src/archive/");
3947        let sibling = RemotePath::new("shared", "bucket", "src-old/");
3948        let other_bucket = RemotePath::new("shared", "other", "src/archive/");
3949
3950        assert!(remote_copy_scopes_overlap(&source, &child));
3951        assert!(remote_copy_scopes_overlap(&child, &source));
3952        assert!(!remote_copy_scopes_overlap(&source, &sibling));
3953        assert!(!remote_copy_scopes_overlap(&source, &other_bucket));
3954    }
3955
3956    #[test]
3957    fn multipart_options_require_and_preserve_planned_source_identity() {
3958        let mut source = rc_core::ObjectInfo::file("large.bin", (MAX_SINGLE_COPY_SIZE + 1) as i64);
3959        source.etag = Some("planned-etag".to_string());
3960        source.version_id = Some("source-version".to_string());
3961        source.content_type = Some("application/octet-stream".to_string());
3962        source.metadata = Some(HashMap::from([(
3963            "project".to_string(),
3964            "archive".to_string(),
3965        )]));
3966
3967        let options = multipart_options_from_source(&source).expect("multipart options");
3968
3969        assert_eq!(options.source_size, MAX_SINGLE_COPY_SIZE + 1);
3970        assert_eq!(options.source_etag, "planned-etag");
3971        assert_eq!(options.source_version_id.as_deref(), Some("source-version"));
3972        assert_eq!(
3973            options.content_type.as_deref(),
3974            Some("application/octet-stream")
3975        );
3976        assert_eq!(
3977            options.metadata.get("project").map(String::as_str),
3978            Some("archive")
3979        );
3980        assert_eq!(
3981            options.metadata.get("rc-source-etag").map(String::as_str),
3982            Some("planned-etag")
3983        );
3984    }
3985
3986    #[test]
3987    fn transfer_age_and_rewind_use_utc_cutoffs() {
3988        let now: Timestamp = "2026-07-21T12:00:00Z".parse().expect("valid UTC timestamp");
3989
3990        assert_eq!(
3991            parse_age_cutoff("1h", now).expect("valid age").to_string(),
3992            "2026-07-21T11:00:00Z"
3993        );
3994        assert_eq!(
3995            parse_rewind_cutoff("2026-07-20T08:30:00+08:00", now)
3996                .expect("valid offset timestamp")
3997                .to_string(),
3998            "2026-07-20T00:30:00Z"
3999        );
4000    }
4001
4002    #[test]
4003    fn transfer_rate_parser_supports_decimal_and_binary_units() {
4004        assert_eq!(parse_byte_rate("10MB/s").expect("decimal rate"), 10_000_000);
4005        assert_eq!(parse_byte_rate("10MiB/s").expect("binary rate"), 10_485_760);
4006        assert!(parse_byte_rate("0").is_err());
4007        assert!(parse_byte_rate("10widgets/s").is_err());
4008    }
4009
4010    #[tokio::test]
4011    async fn planner_rejects_two_sources_that_resolve_to_one_target() {
4012        let (alias_manager, temp_dir) = temp_alias_manager();
4013        let first_dir = temp_dir.path().join("first");
4014        let second_dir = temp_dir.path().join("second");
4015        std::fs::create_dir_all(&first_dir).expect("create first source dir");
4016        std::fs::create_dir_all(&second_dir).expect("create second source dir");
4017        let first = first_dir.join("same.txt");
4018        let second = second_dir.join("same.txt");
4019        std::fs::write(&first, b"first").expect("write first source");
4020        std::fs::write(&second, b"second").expect("write second source");
4021
4022        let candidates = build_transfer_candidates(
4023            &[ParsedPath::Local(first), ParsedPath::Local(second)],
4024            &ParsedPath::Remote(RemotePath::new("target", "bucket", "prefix/")),
4025            true,
4026            false,
4027            None,
4028            None,
4029            &alias_manager,
4030            ObjectKeyPolicy::Logical,
4031        )
4032        .await
4033        .expect("sources can be expanded before selection");
4034        let plan = TransferPlan::build(candidates, &TransferSelection::default());
4035        let error = validate_plan_targets(&plan).expect_err("colliding destinations must fail");
4036
4037        assert!(error.to_string().contains("same destination"));
4038    }
4039
4040    #[test]
4041    fn parse_enc_kms_target_requires_equals_separator() {
4042        let error = parse_kms_target("local/bucket/file.txt").expect_err("missing key separator");
4043        assert!(error.contains("Expected TARGET=KMS_KEY_ID"));
4044    }
4045
4046    #[test]
4047    fn destination_encryption_rejects_local_targets() {
4048        let error = parse_destination_encryption(
4049            &[String::from("./local.txt")],
4050            &[],
4051            &ParsedPath::Local(std::path::PathBuf::from("./local.txt")),
4052        )
4053        .expect_err("local target should be rejected");
4054
4055        assert!(error.contains("must reference a remote destination"));
4056    }
4057
4058    #[test]
4059    fn destination_encryption_detects_conflicting_flags_for_same_target() {
4060        let target = ParsedPath::Remote(RemotePath::new("local", "bucket", "file.txt"));
4061        let error = parse_destination_encryption(
4062            &[String::from("local/bucket/file.txt")],
4063            &[String::from("local/bucket/file.txt=kms-key")],
4064            &target,
4065        )
4066        .expect_err("same target conflict should fail");
4067
4068        assert!(error.contains("cannot target the same destination"));
4069    }
4070
4071    #[test]
4072    fn destination_encryption_rejects_unmatched_s3_target() {
4073        let target = ParsedPath::Remote(RemotePath::new("local", "bucket", "file.txt"));
4074        let error =
4075            parse_destination_encryption(&[String::from("local/bucket/typo.txt")], &[], &target)
4076                .expect_err("unmatched s3 target should fail");
4077
4078        assert!(error.contains("must exactly match the remote destination"));
4079    }
4080
4081    #[test]
4082    fn destination_encryption_rejects_unmatched_kms_target() {
4083        let target = ParsedPath::Remote(RemotePath::new("local", "bucket", "file.txt"));
4084        let error = parse_destination_encryption(
4085            &[],
4086            &[String::from("local/bucket/typo.txt=kms-key")],
4087            &target,
4088        )
4089        .expect_err("unmatched kms target should fail");
4090
4091        assert!(error.contains("must exactly match the remote destination"));
4092    }
4093
4094    #[test]
4095    fn test_cp_output_serialization() {
4096        let output = CpOutput {
4097            status: "success",
4098            source: "src/file.txt".to_string(),
4099            target: "dst/file.txt".to_string(),
4100            size_bytes: Some(1024),
4101            size_human: Some("1 KiB".to_string()),
4102            version_id: Some("destination-v2".to_string()),
4103            source_version_id: Some("source-v1".to_string()),
4104        };
4105        let json = serde_json::to_string(&output).unwrap();
4106        assert!(json.contains("\"status\":\"success\""));
4107        assert!(json.contains("\"size_bytes\":1024"));
4108        assert!(json.contains("\"version_id\":\"destination-v2\""));
4109        assert!(json.contains("\"source_version_id\":\"source-v1\""));
4110    }
4111
4112    #[test]
4113    fn test_cp_output_skips_none_fields() {
4114        let output = CpOutput {
4115            status: "success",
4116            source: "src".to_string(),
4117            target: "dst".to_string(),
4118            size_bytes: None,
4119            size_human: None,
4120            version_id: None,
4121            source_version_id: None,
4122        };
4123        let json = serde_json::to_string(&output).unwrap();
4124        assert!(!json.contains("size_bytes"));
4125        assert!(!json.contains("size_human"));
4126        assert!(!json.contains("version_id"));
4127    }
4128
4129    #[test]
4130    fn versioned_copy_output_uses_v3_and_preserves_both_version_ids() {
4131        let envelope = V3SuccessEnvelope::versioned_objects(VersionCopyData {
4132            operation: "copy",
4133            source: "src/object.txt".to_string(),
4134            target: "dst/object.txt".to_string(),
4135            source_version_id: Some("source-v1".to_string()),
4136            version_id: Some("destination-v2".to_string()),
4137            size_bytes: Some(1024),
4138            size_human: Some("1 KiB".to_string()),
4139        });
4140
4141        let json = serde_json::to_value(envelope).expect("serialize versioned copy output");
4142        assert_eq!(json["schema_version"], 3);
4143        assert_eq!(json["type"], "versioned_objects");
4144        assert_eq!(json["data"]["operation"], "copy");
4145        assert_eq!(json["data"]["source_version_id"], "source-v1");
4146        assert_eq!(json["data"]["version_id"], "destination-v2");
4147    }
4148
4149    #[tokio::test]
4150    async fn storage_class_validation_has_usage_and_unsupported_exit_codes() {
4151        for (storage_class, expected) in [
4152            ("not-a-class", ExitCode::UsageError),
4153            ("STANDARD_IA", ExitCode::UnsupportedFeature),
4154        ] {
4155            let mut args = CpArgs::single("source.txt", "local/bucket/target.txt");
4156            args.storage_class = Some(storage_class.to_string());
4157
4158            assert_eq!(execute(args, OutputConfig::default()).await, expected);
4159        }
4160    }
4161
4162    #[test]
4163    fn fidelity_direction_preflight_rejects_silent_or_beta10_unsupported_paths() {
4164        let local = ParsedPath::Local(PathBuf::from("report.json"));
4165        let source = ParsedPath::Remote(RemotePath::new("test", "source", "report.json"));
4166        let target = ParsedPath::Remote(RemotePath::new("test", "target", "report.json"));
4167
4168        let mut preserve_upload = CpArgs::single("report.json", "test/target/report.json");
4169        preserve_upload.preserve = true;
4170        assert!(matches!(
4171            validate_fidelity_directions(&preserve_upload, std::slice::from_ref(&local), &target),
4172            Err(Error::InvalidPath(_))
4173        ));
4174
4175        let mut replace = CpArgs::single("test/source/report.json", "test/target/report.json");
4176        replace.metadata_directive = Some(MetadataDirectiveArg::Replace);
4177        assert!(matches!(
4178            validate_fidelity_directions(&replace, std::slice::from_ref(&source), &target),
4179            Err(Error::UnsupportedFeature(_))
4180        ));
4181
4182        let cross_alias_source =
4183            ParsedPath::Remote(RemotePath::new("source", "source", "report.json"));
4184        let cross_alias_target =
4185            ParsedPath::Remote(RemotePath::new("destination", "target", "report.json"));
4186        assert!(
4187            validate_fidelity_directions(
4188                &replace,
4189                std::slice::from_ref(&cross_alias_source),
4190                &cross_alias_target,
4191            )
4192            .is_ok(),
4193            "cross-alias metadata REPLACE uses the upload path"
4194        );
4195
4196        let mut tags = CpArgs::single("test/source/report.json", "test/target/report.json");
4197        tags.tagging_directive = Some(TaggingDirectiveArg::Replace);
4198        tags.fidelity.tags = vec!["env=prod".to_string()];
4199        assert!(matches!(
4200            validate_fidelity_directions(&tags, std::slice::from_ref(&source), &target),
4201            Err(Error::UnsupportedFeature(_))
4202        ));
4203
4204        let mut checksum = CpArgs::single("test/source/report.json", "test/target/report.json");
4205        checksum.fidelity.checksum = Some("sha256".to_string());
4206        assert!(matches!(
4207            validate_fidelity_directions(&checksum, std::slice::from_ref(&source), &target),
4208            Err(Error::UnsupportedFeature(_))
4209        ));
4210    }
4211
4212    #[test]
4213    fn source_identity_validation_rejects_same_size_etag_changes() {
4214        let mut planned = ObjectInfo::file("report.json", 4);
4215        planned.etag = Some("planned".to_string());
4216        let mut current = planned.clone();
4217        assert!(source_identity_matches(&planned, &current));
4218
4219        current.etag = Some("changed".to_string());
4220        assert!(!source_identity_matches(&planned, &current));
4221        current.etag = None;
4222        assert!(!source_identity_matches(&planned, &current));
4223
4224        planned.etag = None;
4225        assert!(!source_identity_matches(&planned, &current));
4226    }
4227
4228    #[test]
4229    fn preserve_builds_explicit_metadata_copy_without_replacement_payload() {
4230        let mut args = CpArgs::single("test/source/report.json", "test/target/report.json");
4231        args.preserve = true;
4232        args.fidelity.retention_mode = Some("GOVERNANCE".to_string());
4233        args.fidelity.retain_until = Some("2099-01-02T03:04:05Z".to_string());
4234        args.fidelity.legal_hold = Some("ON".to_string());
4235
4236        let options =
4237            transfer_copy_options(&args, Some("source-v1".to_string()), None).expect("copy policy");
4238        assert_eq!(options.metadata_directive, Some(MetadataDirective::Copy));
4239        assert!(options.destination.attributes.is_none());
4240        assert_eq!(options.source.version_id.as_deref(), Some("source-v1"));
4241        assert!(options.destination.retention.is_some());
4242        assert_eq!(
4243            options.destination.legal_hold,
4244            Some(rc_core::LegalHoldStatus::On)
4245        );
4246    }
4247
4248    #[test]
4249    fn storage_class_plan_rejects_multipart_without_mutation() {
4250        let plan = TransferPlan::build(
4251            vec![TransferCandidate {
4252                payload: CpOperation::RemoteToRemote {
4253                    source: RemotePath::new("local", "source", "large.bin"),
4254                    target: RemotePath::new("local", "target", "large.bin"),
4255                    source_info: Box::new(ObjectInfo::file(
4256                        "large.bin",
4257                        (MAX_SINGLE_COPY_SIZE + 1) as i64,
4258                    )),
4259                    encryption: None,
4260                },
4261                source: "local/source/large.bin".to_string(),
4262                target: "local/target/large.bin".to_string(),
4263                relative_path: "large.bin".to_string(),
4264                modified: None,
4265                size_bytes: Some(MAX_SINGLE_COPY_SIZE + 1),
4266            }],
4267            &TransferSelection::default(),
4268        );
4269
4270        assert!(matches!(
4271            validate_storage_class_plan(&plan, Some("STANDARD")),
4272            Err(Error::UnsupportedFeature(_))
4273        ));
4274    }
4275
4276    #[test]
4277    fn storage_class_plan_rejects_cross_alias_multipart_uploads() {
4278        let plan = TransferPlan::build(
4279            vec![TransferCandidate {
4280                payload: CpOperation::RemoteToRemote {
4281                    source: RemotePath::new("alpha", "source", "medium.bin"),
4282                    target: RemotePath::new("beta", "target", "medium.bin"),
4283                    source_info: Box::new(ObjectInfo::file(
4284                        "medium.bin",
4285                        (MULTIPART_THRESHOLD + 1) as i64,
4286                    )),
4287                    encryption: None,
4288                },
4289                source: "alpha/source/medium.bin".to_string(),
4290                target: "beta/target/medium.bin".to_string(),
4291                relative_path: "medium.bin".to_string(),
4292                modified: None,
4293                size_bytes: Some(MULTIPART_THRESHOLD + 1),
4294            }],
4295            &TransferSelection::default(),
4296        );
4297
4298        assert!(matches!(
4299            validate_storage_class_plan(&plan, Some("STANDARD")),
4300            Err(Error::UnsupportedFeature(_))
4301        ));
4302    }
4303
4304    #[test]
4305    fn piped_copy_preserves_source_content_type_unless_replaced() {
4306        let mut source = ObjectInfo::file("file.txt", 4);
4307        source.content_type = Some("text/plain".to_string());
4308        let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4309
4310        let copied = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4311        assert_eq!(
4312            copied
4313                .attributes
4314                .as_ref()
4315                .and_then(|value| value.content_type.as_deref()),
4316            Some("text/plain")
4317        );
4318
4319        let mut replace = args;
4320        replace.metadata_directive = Some(MetadataDirectiveArg::Replace);
4321        let replaced = piped_copy_write_options(&replace, &source, None).expect("replace metadata");
4322        assert!(
4323            replaced
4324                .attributes
4325                .as_ref()
4326                .is_none_or(|value| value.content_type.is_none())
4327        );
4328    }
4329
4330    #[test]
4331    fn piped_copy_records_the_source_etag_as_identity_metadata() {
4332        let mut source = ObjectInfo::file("file.txt", 4);
4333        source.etag = Some("source-etag".to_string());
4334        let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4335
4336        let options = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4337
4338        let attributes = options.attributes.as_ref().expect("identity attributes");
4339        assert_eq!(
4340            super::super::object_identity::identity_etag_from_metadata(Some(
4341                &attributes.user_metadata
4342            ))
4343            .as_deref(),
4344            Some("source-etag"),
4345            "a later mirror --compare auto must be able to skip this object"
4346        );
4347    }
4348
4349    #[test]
4350    fn piped_copy_records_identity_even_when_metadata_is_replaced() {
4351        let mut source = ObjectInfo::file("file.txt", 4);
4352        source.etag = Some("source-etag".to_string());
4353        source.metadata = Some(HashMap::from([(
4354            "owner".to_string(),
4355            "storage".to_string(),
4356        )]));
4357        let mut args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4358        args.metadata_directive = Some(MetadataDirectiveArg::Replace);
4359
4360        let options = piped_copy_write_options(&args, &source, None).expect("replace metadata");
4361
4362        let attributes = options.attributes.as_ref().expect("identity attributes");
4363        assert_eq!(
4364            super::super::object_identity::identity_etag_from_metadata(Some(
4365                &attributes.user_metadata
4366            ))
4367            .as_deref(),
4368            Some("source-etag"),
4369            "identity is rc bookkeeping, not user metadata"
4370        );
4371        assert!(
4372            !attributes.user_metadata.contains_key("owner"),
4373            "replace must still drop source user metadata"
4374        );
4375    }
4376
4377    #[test]
4378    fn piped_copy_omits_identity_when_the_source_has_no_etag() {
4379        let source = ObjectInfo::file("file.txt", 4);
4380        let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4381
4382        let options = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4383
4384        assert!(
4385            options
4386                .attributes
4387                .as_ref()
4388                .is_none_or(|attributes| attributes.user_metadata.is_empty()),
4389            "without a source ETag there is no identity to record"
4390        );
4391    }
4392
4393    #[test]
4394    fn piped_copy_preserves_source_user_metadata_unless_replaced() {
4395        let mut source = ObjectInfo::file("file.txt", 4);
4396        source.metadata = Some(HashMap::from([(
4397            "owner".to_string(),
4398            "storage".to_string(),
4399        )]));
4400        let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4401
4402        let copied = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4403        assert_eq!(
4404            copied
4405                .attributes
4406                .as_ref()
4407                .and_then(|value| value.user_metadata.get("owner"))
4408                .map(String::as_str),
4409            Some("storage")
4410        );
4411
4412        let mut replace = args;
4413        replace.metadata_directive = Some(MetadataDirectiveArg::Replace);
4414        let replaced = piped_copy_write_options(&replace, &source, None).expect("replace metadata");
4415        assert!(
4416            replaced
4417                .attributes
4418                .as_ref()
4419                .is_none_or(|value| value.user_metadata.is_empty())
4420        );
4421    }
4422
4423    #[test]
4424    fn get_alias_accepts_only_one_remote_source_and_local_target() {
4425        let remote = ParsedPath::Remote(RemotePath::new("local", "reports", "report.json"));
4426        let other_remote = ParsedPath::Remote(RemotePath::new("local", "reports", "other.json"));
4427        let local = ParsedPath::Local(PathBuf::from("./report.json"));
4428
4429        assert_eq!(
4430            validate_alias_direction(TransferAlias::Get, std::slice::from_ref(&remote), &local),
4431            Ok(())
4432        );
4433        assert_eq!(
4434            validate_alias_direction(TransferAlias::Get, &[remote.clone(), other_remote], &local),
4435            Err("get requires exactly one remote source and one local target")
4436        );
4437        assert_eq!(
4438            validate_alias_direction(TransferAlias::Get, std::slice::from_ref(&local), &remote),
4439            Err("get requires exactly one remote source and one local target")
4440        );
4441    }
4442
4443    #[test]
4444    fn put_alias_accepts_only_local_sources_and_remote_target() {
4445        let first_local = ParsedPath::Local(PathBuf::from("./january.csv"));
4446        let second_local = ParsedPath::Local(PathBuf::from("./february.csv"));
4447        let remote = ParsedPath::Remote(RemotePath::new("local", "reports", ""));
4448
4449        assert_eq!(
4450            validate_alias_direction(
4451                TransferAlias::Put,
4452                &[first_local.clone(), second_local],
4453                &remote
4454            ),
4455            Ok(())
4456        );
4457        assert_eq!(
4458            validate_alias_direction(
4459                TransferAlias::Put,
4460                std::slice::from_ref(&remote),
4461                &first_local
4462            ),
4463            Err("put requires one or more local sources and one remote target")
4464        );
4465    }
4466}