rustfs/crates/ecstore/src/bucket/replication/replication_object_config.rs

// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::collections::HashMap;

use rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS;
use s3s::dto::ReplicationConfiguration;
use serde::{Deserialize, Serialize};
use tracing::error;

use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt as _};
use super::replication_error_boundary::Result;
use super::replication_filemeta_boundary::{
    ReplicateDecision, ReplicateTargetDecision, ReplicationStatusType, ReplicationType, ResyncDecision,
};
use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION_RESYNC};
use super::replication_metadata_boundary::ReplicationMetadataStore;
use super::replication_object_decision_boundary::{
    MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision,
    delete_replication_object_opts, delete_replication_version_id, resync_target_for_object,
};
use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, object_to_delete_for_replication};
use super::replication_target_boundary::{BucketTargets, ReplicationTargetStore};
use super::replication_versioning_boundary::ReplicationVersioningStore;
use super::runtime_boundary as runtime_sources;

pub(crate) async fn get_replication_config(bucket: &str) -> Result<Option<ReplicationConfiguration>> {
    ReplicationMetadataStore::optional_replication_config(bucket).await
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ReplicationConfig {
    pub config: Option<ReplicationConfiguration>,
    pub remotes: Option<BucketTargets>,
}

impl ReplicationConfig {
    pub fn new(config: Option<ReplicationConfiguration>, remotes: Option<BucketTargets>) -> Self {
        Self { config, remotes }
    }

    pub fn is_empty(&self) -> bool {
        self.config.is_none()
    }

    pub fn replicate(&self, obj: &ObjectOpts) -> bool {
        self.config.as_ref().is_some_and(|config| config.replicate(obj))
    }

    pub async fn resync(
        &self,
        oi: ObjectInfo,
        dsc: ReplicateDecision,
        status: &HashMap<String, ReplicationStatusType>,
    ) -> ResyncDecision {
        if self.is_empty() {
            return ResyncDecision::default();
        }

        let mut dsc = dsc;

        if oi.delete_marker {
            let opts = ObjectOpts {
                name: oi.name.clone(),
                version_id: delete_replication_version_id(oi.delete_marker, oi.version_id, !oi.version_purge_status.is_empty()),
                delete_marker: true,
                op_type: ReplicationType::Delete,
                existing_object: true,
                ..Default::default()
            };
            let arns = self
                .config
                .as_ref()
                .map(|config| config.filter_target_arns(&opts))
                .unwrap_or_default();

            if arns.is_empty() {
                return ResyncDecision::default();
            }

            for arn in arns {
                let mut opts = opts.clone();
                opts.target_arn = arn;

                dsc.set(ReplicateTargetDecision::new(opts.target_arn.clone(), self.replicate(&opts), false));
            }

            return self.resync_internal(oi, dsc, status);
        }

        let mut user_defined = (*oi.user_defined).clone();
        user_defined.remove(AMZ_BUCKET_REPLICATION_STATUS);

        let dsc = must_replicate(
            oi.bucket.as_str(),
            &oi.name,
            MustReplicateOptions::new(&user_defined, (*oi.user_tags).clone(), ReplicationType::ExistingObject, false),
        )
        .await;

        self.resync_internal(oi, dsc, status)
    }

    fn resync_internal(
        &self,
        oi: ObjectInfo,
        dsc: ReplicateDecision,
        status: &HashMap<String, ReplicationStatusType>,
    ) -> ResyncDecision {
        let Some(remotes) = self.remotes.as_ref() else {
            return ResyncDecision::default();
        };

        if remotes.is_empty() {
            return ResyncDecision::default();
        }

        let mut resync_decision = ResyncDecision::default();

        for target in remotes.targets.iter() {
            if let Some(decision) = dsc.targets_map.get(&target.arn)
                && decision.replicate
            {
                resync_decision.targets.insert(
                    decision.arn.clone(),
                    resync_target_for_object(
                        &ReplicationResyncTargetObject {
                            mod_time: oi.mod_time,
                            user_defined: oi.user_defined.as_ref(),
                        },
                        &target.arn,
                        &target.reset_id,
                        target.reset_before_date,
                        status.get(&decision.arn).unwrap_or(&ReplicationStatusType::Empty).clone(),
                    ),
                );
            }
        }

        resync_decision
    }
}

pub(crate) fn get_must_replicate_options(
    user_defined: &HashMap<String, String>,
    user_tags: String,
    status: ReplicationStatusType,
    op_type: ReplicationType,
    opts: ObjectOptions,
) -> MustReplicateOptions {
    MustReplicateOptions::new(user_defined, user_tags, op_type, opts.replication_request).with_replication_status(status)
}

pub(crate) async fn check_replicate_delete(
    bucket: &str,
    dobj: &ObjectToDelete,
    oi: &ObjectInfo,
    del_opts: &ObjectOptions,
    gerr: Option<String>,
) -> ReplicateDecision {
    let rcfg = match get_replication_config(bucket).await {
        Ok(Some(config)) => config,
        Ok(None) => return ReplicateDecision::default(),
        Err(err) => {
            error!(
                event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED,
                component = LOG_COMPONENT_ECSTORE,
                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
                bucket = %bucket,
                reason = "replication_config_lookup_failed",
                error = %err,
                "Failed to look up replication config for delete replication"
            );
            return ReplicateDecision::default();
        }
    };

    if del_opts.replication_request {
        return ReplicateDecision::default();
    }

    if !del_opts.versioned {
        return ReplicateDecision::default();
    }

    let replication_delete = object_to_delete_for_replication(dobj);
    let opts = delete_replication_object_opts(
        &replication_delete,
        &ReplicationDeleteSource {
            user_defined: oi.user_defined.as_ref(),
            user_tags: oi.user_tags.as_str(),
            delete_marker: oi.delete_marker,
            replication_status: oi.replication_status.clone(),
        },
    );

    let tgt_arns = rcfg.filter_target_arns(&opts);
    let mut dsc = ReplicateDecision::new();

    if tgt_arns.is_empty() {
        return dsc;
    }

    for tgt_arn in tgt_arns {
        let mut opts = opts.clone();
        opts.target_arn = tgt_arn.clone();
        let replicate = rcfg.replicate(&opts);
        let sync = false;

        if gerr.is_some() {
            if let Some(replicate) = delete_replication_missing_source_decision(
                oi.delete_marker,
                oi.target_replication_status(&tgt_arn),
                replicate,
                &oi.version_purge_status,
            ) {
                dsc.set(ReplicateTargetDecision::new(tgt_arn, replicate, sync));
            }
            continue;
        }

        let tgt = ReplicationTargetStore::remote_target_client(bucket, &tgt_arn).await;
        let tgt_dsc = if let Some(tgt) = tgt {
            ReplicateTargetDecision::new(tgt_arn, replicate, tgt.replicate_sync)
        } else {
            ReplicateTargetDecision::new(tgt_arn, false, false)
        };
        dsc.set(tgt_dsc);
    }

    dsc
}

pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision {
    if runtime_sources::object_store_handle().is_none() {
        return ReplicateDecision::default();
    }

    if !ReplicationVersioningStore::prefix_enabled(bucket, object).await {
        return ReplicateDecision::default();
    }

    let replication_status = mopts.replication_status();

    if replication_status == ReplicationStatusType::Replica && !mopts.is_metadata_replication() {
        return ReplicateDecision::default();
    }

    if mopts.is_replication_request() {
        return ReplicateDecision::default();
    }

    let cfg = match get_replication_config(bucket).await {
        Ok(Some(cfg)) => cfg,
        Ok(None) | Err(_) => return ReplicateDecision::default(),
    };

    let opts = ObjectOpts {
        name: object.to_string(),
        replica: replication_status == ReplicationStatusType::Replica,
        existing_object: mopts.is_existing_object_replication(),
        user_tags: mopts.user_tags().to_string(),
        ..Default::default()
    };

    let arns = cfg.filter_target_arns(&opts);

    if arns.is_empty() {
        return ReplicateDecision::default();
    }

    let mut dsc = ReplicateDecision::default();

    for arn in arns {
        let cli = ReplicationTargetStore::remote_target_client(bucket, &arn).await;

        let mut sopts = opts.clone();
        sopts.target_arn = arn.clone();

        let replicate = cfg.replicate(&sopts);
        let synchronous = if let Some(cli) = cli { cli.replicate_sync } else { false };

        dsc.set(ReplicateTargetDecision::new(arn, replicate, synchronous));
    }

    dsc
}

#[cfg(test)]
mod tests {
    use s3s::dto::{
        DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
        ReplicationRule, ReplicationRuleStatus,
    };
    use uuid::Uuid;

    use super::super::replication_filemeta_boundary::VersionPurgeStatusType;
    use super::super::replication_target_boundary::BucketTarget;
    use super::*;

    fn replication_rule() -> ReplicationRule {
        ReplicationRule {
            delete_marker_replication: None,
            delete_replication: None,
            destination: Destination {
                bucket: "arn:aws:s3:::target-bucket".to_string(),
                ..Default::default()
            },
            existing_object_replication: None,
            filter: None,
            id: Some("rule".to_string()),
            prefix: Some(String::new()),
            priority: Some(1),
            source_selection_criteria: None,
            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
        }
    }

    #[test]
    fn replication_config_empty_and_replicate_follow_config() {
        let empty = ReplicationConfig::default();
        assert!(empty.is_empty());
        assert!(!empty.replicate(&ObjectOpts::default()));

        let config = ReplicationConfig::new(
            Some(ReplicationConfiguration {
                role: String::new(),
                rules: vec![replication_rule()],
            }),
            None,
        );

        assert!(!config.is_empty());
        assert!(config.replicate(&ObjectOpts {
            name: "object".to_string(),
            ..Default::default()
        }));
    }

    #[test]
    fn must_replicate_options_preserve_request_flag() {
        let user_defined = HashMap::new();
        let options = get_must_replicate_options(
            &user_defined,
            "env=prod".to_string(),
            ReplicationStatusType::Empty,
            ReplicationType::Metadata,
            ObjectOptions {
                replication_request: true,
                ..Default::default()
            },
        );

        assert!(options.is_metadata_replication());
        assert!(options.is_replication_request());
        assert_eq!(options.user_tags(), "env=prod");
    }

    #[tokio::test]
    async fn resync_distinguishes_delete_marker_creation_from_version_purge() {
        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
        let mut rule = replication_rule();
        rule.destination.bucket = arn.to_string();
        rule.delete_marker_replication = Some(DeleteMarkerReplication {
            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
        });
        rule.delete_replication = Some(DeleteReplication {
            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
        });
        let config = ReplicationConfig::new(
            Some(ReplicationConfiguration {
                role: String::new(),
                rules: vec![rule],
            }),
            Some(BucketTargets {
                targets: vec![BucketTarget {
                    arn: arn.to_string(),
                    endpoint: "target.example".to_string(),
                    ..Default::default()
                }],
            }),
        );
        let marker_version_id = Uuid::new_v4();
        let marker = ObjectInfo {
            bucket: "source".to_string(),
            name: "object".to_string(),
            delete_marker: true,
            version_id: Some(marker_version_id),
            replication_status: ReplicationStatusType::Pending,
            ..Default::default()
        };

        let creation = config
            .resync(marker.clone(), ReplicateDecision::default(), &HashMap::new())
            .await;
        assert!(creation.targets.contains_key(arn));

        let purge = config
            .resync(
                ObjectInfo {
                    version_purge_status: VersionPurgeStatusType::Pending,
                    ..marker
                },
                ReplicateDecision::default(),
                &HashMap::new(),
            )
            .await;
        assert!(purge.targets.is_empty());
    }
}