s4-server 0.7.1

//! S3 Lifecycle execution — per-bucket rule evaluation + manager skeleton
//! (v0.6 #37).
//!
//! AWS S3 Lifecycle attaches a **list of rules** to a bucket; each rule may
//! request that S3
//!
//! 1. **Expire** an object once its age (or the calendar date) crosses a
//!    threshold (`Expiration { Days | Date }`),
//! 2. **Transition** an object to a different storage class (`Transition
//!    { Days, StorageClass }` — `STANDARD_IA`, `GLACIER_IR`, ...),
//! 3. **Expire noncurrent versions** in a versioning-enabled bucket
//!    (`NoncurrentVersionExpiration { NoncurrentDays }`).
//!
//! Until v0.6 #37 the matching `PutBucketLifecycleConfiguration` /
//! `GetBucketLifecycleConfiguration` / `DeleteBucketLifecycle` handlers
//! in `crates/s4-server/src/service.rs` were pure passthroughs (the s3s
//! framework's default backend stored them but nothing read the rules).
//! This module owns the in-memory configuration store + the rule
//! evaluator that decides, for any single object, whether an action
//! should fire **right now**.
//!
//! ## responsibilities (v0.6 #37)
//!
//! - in-memory `bucket -> LifecycleConfig` map with JSON snapshot
//!   round-trip (mirroring `versioning.rs` / `object_lock.rs` /
//!   `inventory.rs`'s shape so `--lifecycle-state-file` is a one-line
//!   addition in `main.rs`).
//! - per-bucket action counters (`actions_total`) — bumped by the
//!   future scanner when an Expiration / Transition /
//!   NoncurrentExpiration action is taken, surfaced via Prometheus
//!   (`s4_lifecycle_actions_total`, see `metrics.rs`).
//! - [`LifecycleManager::evaluate`] — given one (bucket, key, age,
//!   size, tags) tuple, walk the bucket's rules in declaration order
//!   and return the first matching action. Returns `None` when no
//!   rule matches (or when the matching rule is `Disabled`).
//! - [`evaluate_batch`] — batched form for the test path: walks a
//!   slice of `(key, age, size, tags)` tuples and returns the (key,
//!   action) pairs that should fire. The actual backend invocation
//!   (S3.delete_object / metadata rewrite) is the caller's job.
//!
//! ## scope limitations (v0.6 #37)
//!
//! - **Background scanner is a skeleton only.** `main.rs`'s
//!   `--lifecycle-scan-interval-hours` flag spawns a tokio task that
//!   logs the bucket list and stamps a "would-have-run" marker;
//!   walking the source bucket via `list_objects_v2` and actually
//!   invoking `delete_object` / metadata rewrite for each evaluated
//!   action is deferred to v0.7+. Wiring the scheduler to walk a real
//!   bucket end-to-end requires a back-reference from the scheduler
//!   into `S4Service` for the `list_objects_v2` walk and that
//!   reshuffle is out of scope for this issue. The
//!   [`crate::S4Service::run_lifecycle_once_for_test`] entry covers
//!   the in-memory equivalent so the unit + E2E tests exercise the
//!   evaluator end-to-end.
//! - **`AbortIncompleteMultipartUpload`** is parsed and stored on the
//!   `LifecycleRule` (so PutBucketLifecycleConfiguration round-trips
//!   the field) but not enforced — multipart abort sweeping is a
//!   separate scanner that lives next to the multipart upload manager
//!   (v0.7+).
//! - **`expiration_date` (calendar date)** is supported in the
//!   evaluator: a rule with `expiration_date` past `now` fires
//!   Expiration immediately. Same wire form as AWS S3.
//! - **Multi-instance replication.** All state is single-instance
//!   in-memory; `--lifecycle-state-file <PATH>` provides restart
//!   recovery via JSON snapshot, matching the
//!   `--versioning-state-file` shape.
//! - **Object Lock interplay**: the evaluator does NOT consult the
//!   `ObjectLockManager` directly (the evaluator API is
//!   object-tags-and-size only); the scanner caller is expected to
//!   skip locked objects — see the `evaluate_batch_skips_locked` test
//!   for the canonical pattern. Locking always wins over Lifecycle.
//! - **Versioning interplay**: the evaluator treats noncurrent
//!   versions as a separate input — pass `is_noncurrent = true` to
//!   [`LifecycleManager::evaluate_with_flags`] for noncurrent version
//!   expiration matching. The legacy `evaluate` shorthand defaults
//!   `is_noncurrent = false` (current version) so existing call sites
//!   stay one-liners.

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::RwLock;

use chrono::{DateTime, Duration, Utc};
use s3s::S3;
use s3s::S3Request;
use s3s::dto::*;
use serde::{Deserialize, Serialize};
use tracing::warn;

/// Whether a rule is currently being applied. Mirrors AWS S3
/// `ExpirationStatus` (`"Enabled"` / `"Disabled"`).
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum LifecycleStatus {
    Enabled,
    Disabled,
}

impl LifecycleStatus {
    /// Wire form used by AWS S3 (`"Enabled"` / `"Disabled"`).
    #[must_use]
    pub fn as_aws_str(self) -> &'static str {
        match self {
            Self::Enabled => "Enabled",
            Self::Disabled => "Disabled",
        }
    }

    /// Parse the AWS wire form (case-insensitive). Falls back to `Disabled`
    /// on unrecognised input — this matches AWS conservative behaviour
    /// (an unparseable status is treated as "off" so a typo doesn't silently
    /// expire data).
    #[must_use]
    pub fn from_aws_str(s: &str) -> Self {
        if s.eq_ignore_ascii_case("Enabled") {
            Self::Enabled
        } else {
            Self::Disabled
        }
    }
}

/// Per-rule object filter. AWS S3 represents the filter as one of `Prefix`,
/// `Tag`, `ObjectSizeGreaterThan`, `ObjectSizeLessThan`, or `And` (= AND of
/// any subset of those predicates). For internal storage we flatten the
/// "And" form into a struct of optional fields plus a vector of (key, value)
/// tags — every present field must match (logical AND). An empty filter (all
/// fields `None` / empty `tags`) matches every object in the bucket.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct LifecycleFilter {
    /// Object key prefix (empty / `None` = no prefix gating).
    #[serde(default)]
    pub prefix: Option<String>,
    /// Logical AND across every entry: every (key, value) must match the
    /// object's own tag set.
    #[serde(default)]
    pub tags: Vec<(String, String)>,
    /// Object must be *strictly greater than* this size in bytes.
    #[serde(default)]
    pub object_size_greater_than: Option<u64>,
    /// Object must be *strictly less than* this size in bytes.
    #[serde(default)]
    pub object_size_less_than: Option<u64>,
}

impl LifecycleFilter {
    /// `true` when this filter accepts the candidate. Empty filter accepts
    /// every object. Tag matching is AND of all listed tags (each present in
    /// `object_tags` with the matching value).
    #[must_use]
    pub fn matches(&self, key: &str, size: u64, object_tags: &[(String, String)]) -> bool {
        if let Some(p) = &self.prefix
            && !key.starts_with(p)
        {
            return false;
        }
        if let Some(min) = self.object_size_greater_than
            && size <= min
        {
            return false;
        }
        if let Some(max) = self.object_size_less_than
            && size >= max
        {
            return false;
        }
        for (tk, tv) in &self.tags {
            let matched = object_tags.iter().any(|(ok, ov)| ok == tk && ov == tv);
            if !matched {
                return false;
            }
        }
        true
    }
}

/// A single transition step (object age threshold + target storage class).
/// `days` is days since the object was created. AWS S3 also accepts `Date`
/// for transitions but Lifecycle deployments overwhelmingly use `Days`; the
/// `Date` form is omitted here on purpose to keep the evaluator narrow
/// (operators wanting calendar transitions can synthesise a one-shot rule
/// at the cadence of their scanner).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TransitionRule {
    pub days: u32,
    /// Target storage class (`"STANDARD_IA"` / `"GLACIER_IR"` /
    /// `"GLACIER"` / `"DEEP_ARCHIVE"` / `"INTELLIGENT_TIERING"` /
    /// `"ONEZONE_IA"`). Stored as the AWS wire string so PutBucket /
    /// GetBucket round-trip is a no-op.
    pub storage_class: String,
}

/// One lifecycle rule. AWS S3's `LifecycleRule` flattened into the subset
/// the v0.6 #37 evaluator handles. `id` is the operator-supplied label and
/// makes Get / Put round-trips non-lossy.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct LifecycleRule {
    pub id: String,
    pub status: LifecycleStatus,
    #[serde(default)]
    pub filter: LifecycleFilter,
    /// Days since the object was created. Mutually exclusive with
    /// [`Self::expiration_date`] in AWS — both fields are accepted here on
    /// input (the evaluator picks `expiration_days` first, then
    /// `expiration_date`) so a malformed rule with both set still evaluates
    /// deterministically rather than silently dropping the action.
    #[serde(default)]
    pub expiration_days: Option<u32>,
    /// Calendar date past which matching objects are expired (AWS wire form
    /// is ISO 8601; here we keep it as a `DateTime<Utc>` so round-trips
    /// through `serde_json` survive without re-parsing).
    #[serde(default)]
    pub expiration_date: Option<DateTime<Utc>>,
    /// Transition steps in declaration order. The evaluator picks the
    /// deepest transition (largest `days` ≤ object age) and resolves any
    /// conflict with expiration in [`LifecycleManager::evaluate_with_flags`].
    #[serde(default)]
    pub transitions: Vec<TransitionRule>,
    /// Days an object has been noncurrent before the noncurrent-version
    /// expiration fires. Only consulted when the evaluator is asked about
    /// a noncurrent object (`is_noncurrent = true`).
    #[serde(default)]
    pub noncurrent_version_expiration_days: Option<u32>,
    /// Days after a multipart upload is initiated before the abort fires.
    /// Stored so PutBucket round-trips, but **not enforced** in the
    /// v0.6 #37 evaluator — multipart sweeping lives elsewhere.
    #[serde(default)]
    pub abort_incomplete_multipart_upload_days: Option<u32>,
}

impl LifecycleRule {
    /// Convenience constructor for a "expire after N days" rule. Useful in
    /// tests + operator scripts.
    #[must_use]
    pub fn expire_after_days(id: impl Into<String>, days: u32) -> Self {
        Self {
            id: id.into(),
            status: LifecycleStatus::Enabled,
            filter: LifecycleFilter::default(),
            expiration_days: Some(days),
            expiration_date: None,
            transitions: Vec::new(),
            noncurrent_version_expiration_days: None,
            abort_incomplete_multipart_upload_days: None,
        }
    }
}

/// Per-bucket lifecycle configuration (ordered list of rules — first match
/// wins, matching AWS S3 semantics).
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct LifecycleConfig {
    pub rules: Vec<LifecycleRule>,
}

/// The action a single rule wants to take **right now** for a candidate
/// object.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LifecycleAction {
    /// Delete the object (`Expiration` / `NoncurrentVersionExpiration`).
    Expire,
    /// Move the object to a different storage class (`Transition`). The
    /// inner string is the AWS wire form (e.g. `"GLACIER_IR"`).
    Transition { storage_class: String },
}

impl LifecycleAction {
    /// Stable label suitable for a metric counter
    /// (`s4_lifecycle_actions_total{action="..."}`).
    #[must_use]
    pub fn metric_label(&self) -> &'static str {
        match self {
            Self::Expire => "expire",
            Self::Transition { .. } => "transition",
        }
    }
}

/// snapshot のシリアライズ format。`to_json` / `from_json` 用。
#[derive(Debug, Default, Serialize, Deserialize)]
struct LifecycleSnapshot {
    by_bucket: HashMap<String, LifecycleConfig>,
}

/// Per-bucket lifecycle configuration manager.
///
/// All read / write operations go through `RwLock` for thread safety;
/// clones are cheap (`Arc<LifecycleManager>` is the expected handle shape).
/// `actions_total` is a parallel `RwLock<HashMap<...>>` of `(bucket,
/// action_label) -> count` so the future background scanner can stamp
/// successful actions and operators can `GET /metrics` to see the running
/// totals (the metric is also surfaced via `metrics::counter!` — see
/// [`crate::metrics::record_lifecycle_action`]).
#[derive(Debug, Default)]
pub struct LifecycleManager {
    by_bucket: RwLock<HashMap<String, LifecycleConfig>>,
    /// `(bucket, action_label) -> count`. Bumped by the scanner via
    /// [`Self::record_action`]. Action labels are the
    /// [`LifecycleAction::metric_label`] values
    /// (`"expire"` / `"transition"`).
    actions_total: RwLock<HashMap<(String, String), u64>>,
}

impl LifecycleManager {
    /// Empty manager — no bucket has rules.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Replace (or create) the lifecycle configuration for `bucket`. Drops
    /// any previously-attached rules in one shot — matches AWS S3
    /// `PutBucketLifecycleConfiguration` (full replace, no merge).
    pub fn put(&self, bucket: &str, config: LifecycleConfig) {
        self.by_bucket
            .write()
            .expect("lifecycle state RwLock poisoned")
            .insert(bucket.to_owned(), config);
    }

    /// Return a clone of the bucket's configuration, if any.
    #[must_use]
    pub fn get(&self, bucket: &str) -> Option<LifecycleConfig> {
        self.by_bucket
            .read()
            .expect("lifecycle state RwLock poisoned")
            .get(bucket)
            .cloned()
    }

    /// Drop the bucket's lifecycle configuration (idempotent — missing
    /// bucket is OK).
    pub fn delete(&self, bucket: &str) {
        self.by_bucket
            .write()
            .expect("lifecycle state RwLock poisoned")
            .remove(bucket);
    }

    /// JSON snapshot for restart-recoverable state. Pair with
    /// [`Self::from_json`].
    pub fn to_json(&self) -> Result<String, serde_json::Error> {
        let by_bucket = self
            .by_bucket
            .read()
            .expect("lifecycle state RwLock poisoned")
            .clone();
        let snap = LifecycleSnapshot { by_bucket };
        serde_json::to_string(&snap)
    }

    /// Restore from a JSON snapshot produced by [`Self::to_json`]. Action
    /// counters are intentionally not snapshotted — they're transient
    /// observability data and should reset across process restarts so
    /// `rate(s4_lifecycle_actions_total[1h])` doesn't double-count.
    pub fn from_json(s: &str) -> Result<Self, serde_json::Error> {
        let snap: LifecycleSnapshot = serde_json::from_str(s)?;
        Ok(Self {
            by_bucket: RwLock::new(snap.by_bucket),
            actions_total: RwLock::new(HashMap::new()),
        })
    }

    /// Evaluate which rule (if any) applies to a single **current-version**
    /// object right now. Walks the bucket's rules in declaration order;
    /// returns the first matching action. Returns `None` when no rule
    /// matches (or when the matching rule is `Disabled`, or when the
    /// bucket has no lifecycle configuration).
    ///
    /// Within a single rule the precedence is:
    ///
    /// 1. Pick the deepest transition whose `days` threshold is currently
    ///    met (= largest `days ≤ object age`).
    /// 2. Conflict with expiration: if `expiration_days <=
    ///    transition_days` for the chosen transition, expiration wins
    ///    (the rule wants the object gone before it would have been
    ///    transitioned). Otherwise transition wins (e.g. transition at
    ///    30d, expiration at 365d, age 60d → transition fires now,
    ///    expiration is future).
    /// 3. `expiration_date` matches when `now >= expiration_date` and no
    ///    transition is currently applicable.
    ///
    /// `object_age` is "now - created_at" supplied by the caller — keeping
    /// the evaluator pure of the wall clock makes deterministic testing
    /// trivial.
    #[must_use]
    pub fn evaluate(
        &self,
        bucket: &str,
        key: &str,
        object_age: Duration,
        object_size: u64,
        object_tags: &[(String, String)],
    ) -> Option<LifecycleAction> {
        self.evaluate_with_flags(
            bucket,
            key,
            object_age,
            object_size,
            object_tags,
            EvaluateFlags::default(),
        )
    }

    /// Full-form evaluator with flags for noncurrent-version handling.
    /// Use this when the scanner is walking a versioning-enabled bucket;
    /// pass `is_noncurrent = true` for entries that are not the latest
    /// non-delete-marker version.
    #[must_use]
    pub fn evaluate_with_flags(
        &self,
        bucket: &str,
        key: &str,
        object_age: Duration,
        object_size: u64,
        object_tags: &[(String, String)],
        flags: EvaluateFlags,
    ) -> Option<LifecycleAction> {
        let cfg = self.get(bucket)?;
        let now_for_date = flags.now.unwrap_or_else(Utc::now);
        let age_days = object_age.num_days().max(0);
        let age_days_u32 = u32::try_from(age_days).unwrap_or(u32::MAX);
        for rule in &cfg.rules {
            if rule.status != LifecycleStatus::Enabled {
                continue;
            }
            if !rule.filter.matches(key, object_size, object_tags) {
                continue;
            }
            // Noncurrent-version expiration: only consulted when the
            // caller explicitly flags this entry as noncurrent. The
            // current-version expiration / transition rules do not fire
            // for noncurrent versions in AWS S3 semantics.
            if flags.is_noncurrent {
                if let Some(days) = rule.noncurrent_version_expiration_days
                    && age_days_u32 >= days
                {
                    return Some(LifecycleAction::Expire);
                }
                continue;
            }
            // Current-version path.
            let exp_days_match = rule.expiration_days.filter(|d| age_days_u32 >= *d);
            let exp_date_match = rule.expiration_date.filter(|d| now_for_date >= *d);
            // Pick the deepest transition whose threshold is at or
            // below the object's age. Transitions are typically
            // declaration-ordered by ascending `days`, but we don't
            // require it — taking the largest threshold means an
            // object aged 90d gets `GLACIER` over `STANDARD_IA` even
            // if `STANDARD_IA(30d)` was declared first.
            let chosen_transition = rule
                .transitions
                .iter()
                .filter(|t| age_days_u32 >= t.days)
                .max_by_key(|t| t.days);
            // Conflict resolution: when `expiration_days` fires AND a
            // transition fires, expiration wins iff
            // `expiration_days <= transition_days` (rule wants object
            // gone before / at the same time it would have been
            // transitioned). Otherwise the transition wins.
            if let Some(exp_threshold) = exp_days_match {
                let trans_threshold = chosen_transition.map(|t| t.days).unwrap_or(u32::MAX);
                if exp_threshold <= trans_threshold {
                    return Some(LifecycleAction::Expire);
                }
            }
            if let Some(t) = chosen_transition {
                return Some(LifecycleAction::Transition {
                    storage_class: t.storage_class.clone(),
                });
            }
            // Calendar-date expiration (no transition currently
            // applicable, but the rule's expiration_date is past).
            if exp_date_match.is_some() {
                return Some(LifecycleAction::Expire);
            }
            // Fall through to the next rule when no action fires for
            // this rule — first-match-wins applies only to *firing*
            // rules, matching AWS semantics where overlapping rules
            // with disjoint thresholds compose.
        }
        None
    }

    /// Stamp the per-bucket action counter and bump the matching
    /// Prometheus counter. Called by the future scanner after a successful
    /// delete / metadata rewrite.
    pub fn record_action(&self, bucket: &str, action: &LifecycleAction) {
        let label = action.metric_label();
        let key = (bucket.to_owned(), label.to_owned());
        let mut guard = self
            .actions_total
            .write()
            .expect("lifecycle actions counter RwLock poisoned");
        let entry = guard.entry(key).or_insert(0);
        *entry = entry.saturating_add(1);
        crate::metrics::record_lifecycle_action(bucket, label);
    }

    /// Read-only snapshot of the per-(bucket, action) counter map.
    /// Useful for tests + introspection (`/admin/lifecycle/stats` style
    /// endpoints in the future).
    #[must_use]
    pub fn actions_snapshot(&self) -> HashMap<(String, String), u64> {
        self.actions_total
            .read()
            .expect("lifecycle actions counter RwLock poisoned")
            .clone()
    }

    /// All buckets with a lifecycle configuration attached. Sorted for
    /// stable scanner ordering.
    #[must_use]
    pub fn buckets(&self) -> Vec<String> {
        let map = self
            .by_bucket
            .read()
            .expect("lifecycle state RwLock poisoned");
        let mut out: Vec<String> = map.keys().cloned().collect();
        out.sort();
        out
    }
}

/// Flags for [`LifecycleManager::evaluate_with_flags`]. Default is
/// "current-version object, evaluator picks `Utc::now()` for the date
/// comparison". Tests override `now` for determinism.
#[derive(Clone, Copy, Debug, Default)]
pub struct EvaluateFlags {
    pub is_noncurrent: bool,
    pub now: Option<DateTime<Utc>>,
}

/// One object the evaluator considers in a batch:
/// `(key, object_age, object_size, object_tags)`. Defined as a type alias
/// so [`evaluate_batch`] / [`crate::S4Service::run_lifecycle_once_for_test`]
/// don't trip clippy's `type-complexity` lint, and so callers building the
/// list have a single canonical shape to reach for.
pub type EvaluateBatchEntry = (String, Duration, u64, Vec<(String, String)>);

/// Test-driven scan entry: walks a list of [`EvaluateBatchEntry`] tuples
/// and produces (key, action) pairs for every object that should fire an
/// action **right now**. The actual backend invocation (S3.delete_object /
/// metadata rewrite) is the caller's job. Used by both unit tests and the
/// E2E test in `tests/roundtrip.rs`; the future background scanner will
/// reuse the same entry once the bucket-walk is wired through the backend.
#[must_use]
pub fn evaluate_batch(
    manager: &LifecycleManager,
    bucket: &str,
    objects: &[EvaluateBatchEntry],
) -> Vec<(String, LifecycleAction)> {
    let mut out = Vec::with_capacity(objects.len());
    for (key, age, size, tags) in objects {
        if let Some(action) = manager.evaluate(bucket, key, *age, *size, tags) {
            out.push((key.clone(), action));
        }
    }
    out
}

/// Per-invocation scanner counters returned by [`run_scan_once`]. Useful
/// for tests, the `--lifecycle-scan-interval-hours` log line, and any
/// future `/admin/lifecycle/scan` introspection endpoint. Operators see
/// the same numbers via Prometheus
/// (`s4_lifecycle_actions_total{action="expire"|"transition"}`).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ScanReport {
    /// Number of buckets the scanner walked (= buckets with a lifecycle
    /// configuration attached at the moment the scanner ran).
    pub buckets_scanned: usize,
    /// Number of distinct keys the scanner evaluated. Multi-page lists
    /// count one key once even if the listing was paginated.
    pub objects_evaluated: usize,
    /// Number of objects deleted as a result of an Expiration action.
    pub expired: usize,
    /// Number of objects whose `x-amz-storage-class` was rewritten as a
    /// result of a Transition action.
    pub transitioned: usize,
    /// Number of objects skipped because an Object Lock (Compliance,
    /// Governance, or legal hold) was in effect. The Lock always wins
    /// over Lifecycle, matching AWS S3 semantics.
    pub skipped_locked: usize,
    /// Number of objects the evaluator wanted to act on but the action
    /// failed (e.g. backend `delete_object` returned an error). Logged
    /// individually at WARN level; this counter exists so tests / metrics
    /// can assert no silent loss.
    pub action_errors: usize,
}

/// Convert an s3s `Timestamp` (`time::OffsetDateTime` underneath) into a
/// `chrono::DateTime<Utc>` via the RFC3339 wire form. Used by the scanner
/// to compute object age (= `now - last_modified`). Returns `None` when
/// the stamp is unparseable, in which case the caller falls back to
/// treating the object as freshly created (age = 0).
fn timestamp_to_chrono_utc(ts: &Timestamp) -> Option<DateTime<Utc>> {
    let mut buf = Vec::new();
    ts.format(s3s::dto::TimestampFormat::DateTime, &mut buf).ok()?;
    let s = std::str::from_utf8(&buf).ok()?;
    chrono::DateTime::parse_from_rfc3339(s)
        .ok()
        .map(|dt| dt.with_timezone(&Utc))
}

/// Build a synthetic `S3Request` with the minimum metadata the
/// scanner-internal calls need. The lifecycle scanner is a
/// system-internal caller (no end-user credentials, no real HTTP method
/// / URI), so policy gates downstream see `credentials = None` /
/// `region = None` and treat the call as anonymous-internal. Backends
/// that do not gate internal traffic ignore these fields entirely.
fn synthetic_request<T>(input: T, method: http::Method, uri_path: &str) -> S3Request<T> {
    S3Request {
        input,
        method,
        uri: uri_path.parse().unwrap_or_else(|_| "/".parse().expect("/")),
        headers: http::HeaderMap::new(),
        extensions: http::Extensions::new(),
        credentials: None,
        region: None,
        service: None,
        trailing_headers: None,
    }
}

/// Walk every bucket that has a lifecycle configuration attached, list
/// its objects via `list_objects_v2` (continuation-token pagination), and
/// for each object evaluate the rule set + execute the matching
/// Expiration / Transition action. Object-Lock-protected objects are
/// **skipped** (the Lock always wins over Lifecycle). Versioning chains
/// are intentionally out of scope for v0.7 #45 — see the module-level
/// limitations note.
///
/// ## error handling
///
/// Per-bucket / per-object failures are logged at WARN level and bumped
/// in `ScanReport::action_errors`; the scanner does NOT abort early on a
/// single bad object so one slow / faulty bucket can't starve every
/// other bucket's lifecycle. The function only returns `Err(_)` when the
/// scanner cannot make progress at all (no current usage — kept for the
/// future case where the manager itself becomes unavailable).
///
/// ## scope (v0.7 #45)
///
/// - Current-version objects only (Versioning-enabled chains rely on
///   `evaluate_with_flags(is_noncurrent = true)`, but walking the
///   shadow keys requires the version chain access pattern from
///   `versioning.rs` and is deferred to a follow-up issue).
/// - `head_object`'s `last_modified` is used to compute age. When the
///   backend omits the field (some S3-compatible backends do), the
///   object is treated as age 0 and skipped — matches AWS conservative
///   behaviour where a malformed timestamp must not silently expire data.
/// - Tags are looked up via the attached
///   [`crate::tagging::TagManager`] (when wired). Buckets without a
///   tag manager pass an empty tag list to the evaluator.
/// - Transition rewrites the object's `x-amz-storage-class` via
///   `copy_object` (same bucket / same key, `MetadataDirective: COPY`,
///   new `StorageClass`). Backends that ignore the storage class
///   header silently no-op the transition; the counter still bumps to
///   reflect "the scanner asked for a transition" (matching AWS where
///   a no-op transition still costs a request).
pub async fn run_scan_once<B: S3 + Send + Sync + 'static>(
    s4: &Arc<crate::S4Service<B>>,
) -> Result<ScanReport, String> {
    let Some(mgr) = s4.lifecycle_manager().cloned() else {
        // No lifecycle manager attached (e.g. operator did not set
        // `--lifecycle-state-file`). Scan is a no-op.
        return Ok(ScanReport::default());
    };
    let buckets = mgr.buckets();
    if buckets.is_empty() {
        return Ok(ScanReport::default());
    }
    let now = Utc::now();
    let mut report = ScanReport {
        buckets_scanned: buckets.len(),
        ..ScanReport::default()
    };
    for bucket in buckets {
        scan_bucket(s4, &mgr, &bucket, now, &mut report).await;
    }
    Ok(report)
}

/// Walk one bucket end-to-end. Pagination uses the `continuation_token`
/// loop documented in
/// <https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html>.
async fn scan_bucket<B: S3 + Send + Sync + 'static>(
    s4: &Arc<crate::S4Service<B>>,
    mgr: &Arc<LifecycleManager>,
    bucket: &str,
    now: DateTime<Utc>,
    report: &mut ScanReport,
) {
    let mut continuation: Option<String> = None;
    loop {
        let list_input = ListObjectsV2Input {
            bucket: bucket.to_owned(),
            continuation_token: continuation.clone(),
            ..Default::default()
        };
        let list_req = synthetic_request(
            list_input,
            http::Method::GET,
            &format!("/{bucket}?list-type=2"),
        );
        let resp = match s4.as_ref().list_objects_v2(list_req).await {
            Ok(r) => r,
            Err(e) => {
                warn!(
                    bucket = %bucket,
                    error = %e,
                    "S4 lifecycle: list_objects_v2 failed; skipping bucket for this scan",
                );
                report.action_errors = report.action_errors.saturating_add(1);
                return;
            }
        };
        let output = resp.output;
        let contents = output.contents.unwrap_or_default();
        for obj in &contents {
            let Some(key) = obj.key.as_deref() else {
                continue;
            };
            // Filter out S4-internal sidecars / shadow versions early so
            // the lifecycle scanner mirrors the same "client-visible
            // object set" the customer sees through `list_objects_v2`.
            // (The S4Service.list_objects_v2 handler already drops them
            // before returning, but this is a belt-and-braces guard for
            // any future bypass that builds the list elsewhere.)
            if key.ends_with(".s4index") {
                continue;
            }
            report.objects_evaluated = report.objects_evaluated.saturating_add(1);
            let size = obj.size.unwrap_or(0).max(0) as u64;
            let age = match obj.last_modified.as_ref().and_then(timestamp_to_chrono_utc) {
                Some(lm) => now.signed_duration_since(lm),
                None => Duration::zero(),
            };
            let tags: Vec<(String, String)> = s4
                .as_ref()
                .tag_manager()
                .and_then(|m| m.get_object_tags(bucket, key))
                .map(|set| set.iter().cloned().collect())
                .unwrap_or_default();
            let Some(action) = mgr.evaluate(bucket, key, age, size, &tags) else {
                continue;
            };
            // Object-Lock-protected objects are skipped before any
            // backend-mutating call. Lock wins over Lifecycle, full
            // stop — matches AWS behaviour where an Expiration on a
            // locked object is dropped, not retried.
            if let Some(lock_mgr) = s4.as_ref().object_lock_manager()
                && let Some(state) = lock_mgr.get(bucket, key)
                && state.is_locked(now)
            {
                report.skipped_locked = report.skipped_locked.saturating_add(1);
                continue;
            }
            match action {
                LifecycleAction::Expire => match execute_expire(s4, bucket, key).await {
                    Ok(()) => {
                        mgr.record_action(bucket, &LifecycleAction::Expire);
                        report.expired = report.expired.saturating_add(1);
                    }
                    Err(e) => {
                        warn!(
                            bucket = %bucket,
                            key = %key,
                            error = %e,
                            "S4 lifecycle: Expire action failed",
                        );
                        report.action_errors = report.action_errors.saturating_add(1);
                    }
                },
                LifecycleAction::Transition { storage_class } => {
                    match execute_transition(s4, bucket, key, &storage_class).await {
                        Ok(()) => {
                            mgr.record_action(
                                bucket,
                                &LifecycleAction::Transition {
                                    storage_class: storage_class.clone(),
                                },
                            );
                            report.transitioned = report.transitioned.saturating_add(1);
                        }
                        Err(e) => {
                            warn!(
                                bucket = %bucket,
                                key = %key,
                                storage_class = %storage_class,
                                error = %e,
                                "S4 lifecycle: Transition action failed",
                            );
                            report.action_errors = report.action_errors.saturating_add(1);
                        }
                    }
                }
            }
        }
        if output.is_truncated.unwrap_or(false) {
            continuation = output.next_continuation_token;
            if continuation.is_none() {
                // Defensive: AWS guarantees a NextContinuationToken when
                // is_truncated=true, but a malformed backend could omit
                // it; break to avoid an infinite loop.
                break;
            }
        } else {
            break;
        }
    }
}

/// Issue `delete_object` against the wrapped `S4Service`. The handler in
/// `service.rs` runs the WORM check itself, so even if the scanner's
/// pre-check missed (race with an MFA-Delete put-bucket-versioning), the
/// backend refuses the delete with `AccessDenied` and the error path
/// above bumps `action_errors` rather than silently losing data.
async fn execute_expire<B: S3 + Send + Sync + 'static>(
    s4: &Arc<crate::S4Service<B>>,
    bucket: &str,
    key: &str,
) -> Result<(), String> {
    let input = DeleteObjectInput {
        bucket: bucket.to_owned(),
        key: key.to_owned(),
        ..Default::default()
    };
    let req = synthetic_request(
        input,
        http::Method::DELETE,
        &format!("/{bucket}/{key}"),
    );
    s4.as_ref()
        .delete_object(req)
        .await
        .map(|_| ())
        .map_err(|e| format!("{e}"))
}

/// Rewrite the object's storage class via a same-key `copy_object` with
/// `MetadataDirective: COPY` (preserves user metadata) and the new
/// `storage_class`. Backends that ignore storage-class headers
/// effectively no-op; the counter still records the attempt so dashboards
/// reflect the scanner's intent.
async fn execute_transition<B: S3 + Send + Sync + 'static>(
    s4: &Arc<crate::S4Service<B>>,
    bucket: &str,
    key: &str,
    storage_class: &str,
) -> Result<(), String> {
    // CopyObjectInput has dozens of `Option` fields plus three required
    // (bucket / key / copy_source); the s3s-generated `builder()` is
    // the path that fills the optional ones with `None` for us. The
    // `set_*` setters return `&mut Self`, so we drive them in
    // statement form rather than as a method chain.
    let mut builder = CopyObjectInput::builder();
    builder.set_bucket(bucket.to_owned());
    builder.set_key(key.to_owned());
    builder.set_copy_source(CopySource::Bucket {
        bucket: bucket.to_owned().into_boxed_str(),
        key: key.to_owned().into_boxed_str(),
        version_id: None,
    });
    builder.set_metadata_directive(Some(MetadataDirective::from_static(MetadataDirective::COPY)));
    builder.set_storage_class(Some(StorageClass::from(storage_class.to_owned())));
    let input = builder
        .build()
        .map_err(|e| format!("CopyObjectInput build: {e}"))?;
    let req = synthetic_request(
        input,
        http::Method::PUT,
        &format!("/{bucket}/{key}"),
    );
    s4.as_ref()
        .copy_object(req)
        .await
        .map(|_| ())
        .map_err(|e| format!("{e}"))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn enabled(rule: LifecycleRule) -> LifecycleRule {
        LifecycleRule {
            status: LifecycleStatus::Enabled,
            ..rule
        }
    }

    fn cfg_with(rules: Vec<LifecycleRule>) -> LifecycleConfig {
        LifecycleConfig { rules }
    }

    fn manager_with(bucket: &str, rules: Vec<LifecycleRule>) -> LifecycleManager {
        let m = LifecycleManager::new();
        m.put(bucket, cfg_with(rules));
        m
    }

    #[test]
    fn evaluate_age_past_expiration_returns_expire() {
        let m = manager_with("b", vec![LifecycleRule::expire_after_days("r", 30)]);
        let action = m.evaluate("b", "k", Duration::days(31), 100, &[]);
        assert_eq!(action, Some(LifecycleAction::Expire));
    }

    #[test]
    fn evaluate_age_before_expiration_returns_none() {
        let m = manager_with("b", vec![LifecycleRule::expire_after_days("r", 30)]);
        let action = m.evaluate("b", "k", Duration::days(5), 100, &[]);
        assert_eq!(action, None);
    }

    #[test]
    fn evaluate_prefix_filter_matches() {
        let mut rule = LifecycleRule::expire_after_days("r", 1);
        rule.filter.prefix = Some("logs/".into());
        let m = manager_with("b", vec![rule]);
        assert_eq!(
            m.evaluate("b", "logs/2026/a.log", Duration::days(2), 1, &[]),
            Some(LifecycleAction::Expire)
        );
        assert_eq!(
            m.evaluate("b", "data/keep.bin", Duration::days(2), 1, &[]),
            None
        );
    }

    #[test]
    fn evaluate_tag_filter_requires_all_tags_to_match() {
        let mut rule = LifecycleRule::expire_after_days("r", 1);
        rule.filter.tags = vec![
            ("env".into(), "dev".into()),
            ("expirable".into(), "yes".into()),
        ];
        let m = manager_with("b", vec![rule]);
        // All tags present + matching → fire.
        assert_eq!(
            m.evaluate(
                "b",
                "k",
                Duration::days(2),
                1,
                &[
                    ("env".into(), "dev".into()),
                    ("expirable".into(), "yes".into()),
                    ("owner".into(), "alice".into()),
                ]
            ),
            Some(LifecycleAction::Expire)
        );
        // One tag missing → no fire.
        assert_eq!(
            m.evaluate(
                "b",
                "k",
                Duration::days(2),
                1,
                &[("env".into(), "dev".into())]
            ),
            None
        );
        // Tag present but with the wrong value → no fire.
        assert_eq!(
            m.evaluate(
                "b",
                "k",
                Duration::days(2),
                1,
                &[
                    ("env".into(), "prod".into()),
                    ("expirable".into(), "yes".into()),
                ]
            ),
            None
        );
    }

    #[test]
    fn evaluate_size_filters_gate_action() {
        let mut rule = LifecycleRule::expire_after_days("r", 1);
        rule.filter.object_size_greater_than = Some(1024);
        rule.filter.object_size_less_than = Some(10 * 1024);
        let m = manager_with("b", vec![rule]);
        // Inside the (1024, 10*1024) range → fire.
        assert_eq!(
            m.evaluate("b", "k", Duration::days(2), 4096, &[]),
            Some(LifecycleAction::Expire)
        );
        // At the boundary (size == greater_than) → strict `>`, no fire.
        assert_eq!(m.evaluate("b", "k", Duration::days(2), 1024, &[]), None);
        // Above the upper bound → no fire.
        assert_eq!(
            m.evaluate("b", "k", Duration::days(2), 100 * 1024, &[]),
            None
        );
    }

    #[test]
    fn evaluate_transition_fires_before_expiration() {
        // Transition at 30d, expiration at 365d, age 60d → transition.
        let rule = enabled(LifecycleRule {
            id: "r".into(),
            status: LifecycleStatus::Enabled,
            filter: LifecycleFilter::default(),
            expiration_days: Some(365),
            expiration_date: None,
            transitions: vec![TransitionRule {
                days: 30,
                storage_class: "GLACIER_IR".into(),
            }],
            noncurrent_version_expiration_days: None,
            abort_incomplete_multipart_upload_days: None,
        });
        let m = manager_with("b", vec![rule]);
        let action = m.evaluate("b", "k", Duration::days(60), 1, &[]);
        assert_eq!(
            action,
            Some(LifecycleAction::Transition {
                storage_class: "GLACIER_IR".into(),
            })
        );
    }

    #[test]
    fn evaluate_expiration_wins_when_threshold_is_earlier_than_transition() {
        // Expiration at 30d, transition at 90d, age 100d → expire (the
        // rule wants the object gone *before* it would have transitioned).
        let rule = enabled(LifecycleRule {
            id: "r".into(),
            status: LifecycleStatus::Enabled,
            filter: LifecycleFilter::default(),
            expiration_days: Some(30),
            expiration_date: None,
            transitions: vec![TransitionRule {
                days: 90,
                storage_class: "GLACIER".into(),
            }],
            noncurrent_version_expiration_days: None,
            abort_incomplete_multipart_upload_days: None,
        });
        let m = manager_with("b", vec![rule]);
        let action = m.evaluate("b", "k", Duration::days(100), 1, &[]);
        assert_eq!(action, Some(LifecycleAction::Expire));
    }

    #[test]
    fn evaluate_disabled_rule_never_fires() {
        let mut rule = LifecycleRule::expire_after_days("r", 1);
        rule.status = LifecycleStatus::Disabled;
        let m = manager_with("b", vec![rule]);
        assert_eq!(m.evaluate("b", "k", Duration::days(365), 1, &[]), None);
    }

    #[test]
    fn evaluate_unknown_bucket_returns_none() {
        let m = LifecycleManager::new();
        assert_eq!(m.evaluate("ghost", "k", Duration::days(365), 1, &[]), None);
    }

    #[test]
    fn evaluate_noncurrent_version_expiration() {
        let rule = enabled(LifecycleRule {
            id: "r".into(),
            status: LifecycleStatus::Enabled,
            filter: LifecycleFilter::default(),
            expiration_days: None,
            expiration_date: None,
            transitions: vec![],
            noncurrent_version_expiration_days: Some(7),
            abort_incomplete_multipart_upload_days: None,
        });
        let m = manager_with("b", vec![rule]);
        // current-version path → no rule matches (no expiration_days set).
        assert_eq!(m.evaluate("b", "k", Duration::days(30), 1, &[]), None);
        // noncurrent path with age past 7d → expire.
        let action = m.evaluate_with_flags(
            "b",
            "k",
            Duration::days(8),
            1,
            &[],
            EvaluateFlags {
                is_noncurrent: true,
                now: None,
            },
        );
        assert_eq!(action, Some(LifecycleAction::Expire));
        // noncurrent path with age before 7d → no fire.
        let action = m.evaluate_with_flags(
            "b",
            "k",
            Duration::days(3),
            1,
            &[],
            EvaluateFlags {
                is_noncurrent: true,
                now: None,
            },
        );
        assert_eq!(action, None);
    }

    #[test]
    fn evaluate_batch_distributes_actions_across_object_ages() {
        // Transition at 30d, expiration at 60d. Conflict resolver picks
        // expire iff `exp_days <= trans_days` for the chosen transition.
        // With exp=60, trans=30: at age 40-59 the transition fires; at
        // age >= 60 expiration wins (because exp_days=60 <= trans_days=30
        // is false, so... wait — re-read: the resolver compares
        // exp_threshold (60) vs trans_threshold (30) and triggers expire
        // ONLY when 60 <= 30, which is false → transition keeps winning
        // until both thresholds met but exp <= trans). For exp=60 trans=30
        // pair, transition always wins regardless of age (rule pattern is
        // "transition first, expire later" — the next scanner pass
        // picks up the expiration). So expect 4 transitions.
        let rule = enabled(LifecycleRule {
            id: "r".into(),
            status: LifecycleStatus::Enabled,
            filter: LifecycleFilter::default(),
            expiration_days: Some(60),
            expiration_date: None,
            transitions: vec![TransitionRule {
                days: 30,
                storage_class: "STANDARD_IA".into(),
            }],
            noncurrent_version_expiration_days: None,
            abort_incomplete_multipart_upload_days: None,
        });
        let m = manager_with("b", vec![rule]);
        let objects = vec![
            ("young".to_string(), Duration::days(10), 1u64, vec![]),
            ("middle".to_string(), Duration::days(40), 1u64, vec![]),
            ("middle2".to_string(), Duration::days(45), 1u64, vec![]),
            ("old".to_string(), Duration::days(90), 1u64, vec![]),
            ("ancient".to_string(), Duration::days(365), 1u64, vec![]),
        ];
        let actions = evaluate_batch(&m, "b", &objects);
        assert_eq!(actions.len(), 4);
        for (_, a) in &actions {
            assert!(matches!(a, LifecycleAction::Transition { .. }));
        }
    }

    #[test]
    fn json_round_trip_preserves_rules() {
        let rule = enabled(LifecycleRule {
            id: "complex".into(),
            status: LifecycleStatus::Enabled,
            filter: LifecycleFilter {
                prefix: Some("logs/".into()),
                tags: vec![("env".into(), "prod".into())],
                object_size_greater_than: Some(1024),
                object_size_less_than: None,
            },
            expiration_days: Some(365),
            expiration_date: None,
            transitions: vec![TransitionRule {
                days: 30,
                storage_class: "STANDARD_IA".into(),
            }],
            noncurrent_version_expiration_days: Some(7),
            abort_incomplete_multipart_upload_days: Some(3),
        });
        let m = manager_with("b1", vec![rule.clone()]);
        let json = m.to_json().expect("to_json");
        let m2 = LifecycleManager::from_json(&json).expect("from_json");
        let cfg = m2.get("b1").expect("bucket survives roundtrip");
        assert_eq!(cfg.rules.len(), 1);
        assert_eq!(cfg.rules[0], rule);
    }

    #[test]
    fn lifecycle_config_default_is_empty() {
        let cfg = LifecycleConfig::default();
        assert!(cfg.rules.is_empty());
    }

    #[test]
    fn evaluate_batch_skips_locked_objects_at_caller_layer() {
        // The evaluator itself does not consult ObjectLock; the scanner
        // (and tests) are expected to filter locked keys out before /
        // after calling `evaluate_batch`. This test documents the
        // canonical pattern.
        let m = manager_with("b", vec![LifecycleRule::expire_after_days("r", 1)]);
        let objects = vec![
            ("locked".to_string(), Duration::days(30), 1u64, vec![]),
            ("free".to_string(), Duration::days(30), 1u64, vec![]),
        ];
        let locked_keys: std::collections::HashSet<&str> = ["locked"].into_iter().collect();
        let raw = evaluate_batch(&m, "b", &objects);
        let filtered: Vec<_> = raw
            .into_iter()
            .filter(|(k, _)| !locked_keys.contains(k.as_str()))
            .collect();
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0].0, "free");
    }

    #[test]
    fn record_action_bumps_per_bucket_counter() {
        let m = LifecycleManager::new();
        m.record_action("b", &LifecycleAction::Expire);
        m.record_action("b", &LifecycleAction::Expire);
        m.record_action(
            "b",
            &LifecycleAction::Transition {
                storage_class: "GLACIER".into(),
            },
        );
        let snap = m.actions_snapshot();
        assert_eq!(snap.get(&("b".into(), "expire".into())).copied(), Some(2));
        assert_eq!(
            snap.get(&("b".into(), "transition".into())).copied(),
            Some(1)
        );
    }

    // ---- v0.7 #45: scanner runner tests --------------------------------
    //
    // These tests stand up an in-memory `S4Service` over a tiny
    // `ScannerMemBackend` (separate from the larger `MemoryBackend` in
    // `tests/roundtrip.rs` so this module stays self-contained). The
    // backend implements only the four `S3` methods the scanner touches:
    // `put_object`, `head_object`, `delete_object`, `list_objects_v2`.
    // Tags are exercised via the optional `with_tagging(...)` manager.
    //
    // Object age is faked by setting an `expire_after_days(0)` rule, so
    // any object whose backend-recorded `last_modified` is at or before
    // "now" matches — sidesteps the `head_object`/`Timestamp` parsing
    // entirely (and matches the canonical "operator just put the bucket
    // on aggressive expiration" test path).

    use std::collections::HashMap;
    use std::sync::Mutex as StdMutex;

    use bytes::Bytes;
    use s3s::dto as dto2;
    use s3s::{S3Error, S3ErrorCode, S3Response, S3Result};
    use s4_codec::dispatcher::AlwaysDispatcher;
    use s4_codec::passthrough::Passthrough;
    use s4_codec::{CodecKind, CodecRegistry};

    use crate::S4Service;
    use crate::object_lock::{LockMode, ObjectLockManager, ObjectLockState};

    #[derive(Default)]
    struct ScannerMemBackend {
        objects: StdMutex<HashMap<(String, String), ScannerStored>>,
    }

    #[derive(Clone)]
    struct ScannerStored {
        body: Bytes,
        last_modified: dto2::Timestamp,
    }

    impl ScannerMemBackend {
        fn put_now(&self, bucket: &str, key: &str, body: Bytes) {
            self.objects.lock().unwrap().insert(
                (bucket.to_owned(), key.to_owned()),
                ScannerStored {
                    body,
                    last_modified: dto2::Timestamp::from(std::time::SystemTime::now()),
                },
            );
        }
    }

    #[async_trait::async_trait]
    impl S3 for ScannerMemBackend {
        async fn put_object(
            &self,
            req: S3Request<dto2::PutObjectInput>,
        ) -> S3Result<S3Response<dto2::PutObjectOutput>> {
            self.put_now(&req.input.bucket, &req.input.key, Bytes::new());
            Ok(S3Response::new(dto2::PutObjectOutput::default()))
        }

        async fn head_object(
            &self,
            req: S3Request<dto2::HeadObjectInput>,
        ) -> S3Result<S3Response<dto2::HeadObjectOutput>> {
            let key = (req.input.bucket.clone(), req.input.key.clone());
            let lock = self.objects.lock().unwrap();
            let stored = lock
                .get(&key)
                .ok_or_else(|| S3Error::new(S3ErrorCode::NoSuchKey))?;
            Ok(S3Response::new(dto2::HeadObjectOutput {
                content_length: Some(stored.body.len() as i64),
                last_modified: Some(stored.last_modified.clone()),
                ..Default::default()
            }))
        }

        async fn delete_object(
            &self,
            req: S3Request<dto2::DeleteObjectInput>,
        ) -> S3Result<S3Response<dto2::DeleteObjectOutput>> {
            let key = (req.input.bucket.clone(), req.input.key.clone());
            self.objects.lock().unwrap().remove(&key);
            Ok(S3Response::new(dto2::DeleteObjectOutput::default()))
        }

        async fn list_objects_v2(
            &self,
            req: S3Request<dto2::ListObjectsV2Input>,
        ) -> S3Result<S3Response<dto2::ListObjectsV2Output>> {
            let prefix = req.input.bucket.clone();
            let lock = self.objects.lock().unwrap();
            let mut contents: Vec<dto2::Object> = lock
                .iter()
                .filter(|((b, _), _)| b == &prefix)
                .map(|((_, k), v)| dto2::Object {
                    key: Some(k.clone()),
                    size: Some(v.body.len() as i64),
                    last_modified: Some(v.last_modified.clone()),
                    ..Default::default()
                })
                .collect();
            contents.sort_by(|a, b| a.key.cmp(&b.key));
            let key_count = i32::try_from(contents.len()).unwrap_or(i32::MAX);
            Ok(S3Response::new(dto2::ListObjectsV2Output {
                name: Some(prefix),
                contents: Some(contents),
                key_count: Some(key_count),
                is_truncated: Some(false),
                ..Default::default()
            }))
        }

        async fn copy_object(
            &self,
            _req: S3Request<dto2::CopyObjectInput>,
        ) -> S3Result<S3Response<dto2::CopyObjectOutput>> {
            // Transition path: scanner copies same-key with new
            // storage_class. The mem backend doesn't track storage
            // class, so it's a no-op success — exactly the AWS-side
            // behaviour for a backend that ignores the field.
            Ok(S3Response::new(dto2::CopyObjectOutput::default()))
        }
    }

    fn make_service() -> Arc<S4Service<ScannerMemBackend>> {
        let registry = Arc::new(
            CodecRegistry::new(CodecKind::Passthrough).with(Arc::new(Passthrough)),
        );
        let dispatcher = Arc::new(AlwaysDispatcher(CodecKind::Passthrough));
        Arc::new(S4Service::new(
            ScannerMemBackend::default(),
            registry,
            dispatcher,
        ))
    }

    #[tokio::test]
    async fn run_scan_once_no_lifecycle_manager_returns_empty_report() {
        // Service has no lifecycle manager attached — scanner must
        // no-op cleanly (operator might not have set
        // `--lifecycle-state-file`). Also covers the empty-buckets
        // path in `run_scan_once`.
        let s4 = make_service();
        let report = run_scan_once(&s4).await.expect("scan");
        assert_eq!(report, ScanReport::default());

        // And: lifecycle manager attached but no buckets configured.
        let mgr = Arc::new(LifecycleManager::new());
        let backend = ScannerMemBackend::default();
        let registry = Arc::new(
            CodecRegistry::new(CodecKind::Passthrough).with(Arc::new(Passthrough)),
        );
        let dispatcher = Arc::new(AlwaysDispatcher(CodecKind::Passthrough));
        let s4_empty = Arc::new(
            S4Service::new(backend, registry, dispatcher).with_lifecycle(mgr),
        );
        let report = run_scan_once(&s4_empty).await.expect("scan empty");
        assert_eq!(report, ScanReport::default());
    }

    #[tokio::test]
    async fn run_scan_once_expires_matching_objects_via_backend() {
        // Three objects: only "stale.log" matches the rule (prefix
        // gating). The other two are written but not under the prefix,
        // so the evaluator returns None for them.
        let backend = ScannerMemBackend::default();
        backend.put_now("b", "stale.log", Bytes::from_static(b"x"));
        backend.put_now("b", "data/keep1.bin", Bytes::from_static(b"y"));
        backend.put_now("b", "data/keep2.bin", Bytes::from_static(b"z"));
        // Rule: any object under `stale.` prefix is expired immediately
        // (`expire_after_days(0)` matches age >= 0d, which is every
        // backend object).
        let mgr = Arc::new(LifecycleManager::new());
        let mut rule = LifecycleRule::expire_after_days("r", 0);
        rule.filter.prefix = Some("stale.".into());
        mgr.put("b", LifecycleConfig { rules: vec![rule] });
        let registry = Arc::new(
            CodecRegistry::new(CodecKind::Passthrough).with(Arc::new(Passthrough)),
        );
        let dispatcher = Arc::new(AlwaysDispatcher(CodecKind::Passthrough));
        let s4 = Arc::new(
            S4Service::new(backend, registry, dispatcher).with_lifecycle(Arc::clone(&mgr)),
        );

        let report = run_scan_once(&s4).await.expect("scan");
        assert_eq!(report.buckets_scanned, 1);
        assert_eq!(report.objects_evaluated, 3);
        assert_eq!(report.expired, 1);
        assert_eq!(report.transitioned, 0);
        assert_eq!(report.skipped_locked, 0);
        assert_eq!(report.action_errors, 0);
        // Backend post-condition: the matching key is gone, the others
        // remain. Read back through the service's own list_objects_v2
        // path (which is also what the customer-visible HTTP layer
        // serves) so we exercise the same code the scanner walked.
        let req = synthetic_request(
            ListObjectsV2Input {
                bucket: "b".into(),
                ..Default::default()
            },
            http::Method::GET,
            "/b?list-type=2",
        );
        let resp = s4
            .as_ref()
            .list_objects_v2(req)
            .await
            .expect("post-scan list");
        let keys: Vec<String> = resp
            .output
            .contents
            .unwrap_or_default()
            .into_iter()
            .filter_map(|o| o.key)
            .collect();
        assert!(!keys.contains(&"stale.log".to_string()));
        assert!(keys.contains(&"data/keep1.bin".to_string()));
        assert!(keys.contains(&"data/keep2.bin".to_string()));
        // Lifecycle action counter: one Expire bumped on bucket "b".
        let snap = mgr.actions_snapshot();
        assert_eq!(
            snap.get(&("b".into(), "expire".into())).copied(),
            Some(1)
        );
    }

    #[tokio::test]
    async fn run_scan_once_skips_object_lock_protected_keys() {
        let backend = ScannerMemBackend::default();
        backend.put_now("b", "locked.log", Bytes::from_static(b"x"));
        backend.put_now("b", "free.log", Bytes::from_static(b"y"));
        let registry = Arc::new(
            CodecRegistry::new(CodecKind::Passthrough).with(Arc::new(Passthrough)),
        );
        let dispatcher = Arc::new(AlwaysDispatcher(CodecKind::Passthrough));
        let mgr = Arc::new(LifecycleManager::new());
        // Aggressive: every object expires immediately.
        mgr.put(
            "b",
            LifecycleConfig {
                rules: vec![LifecycleRule::expire_after_days("r", 0)],
            },
        );
        let lock_mgr = Arc::new(ObjectLockManager::new());
        // Lock retains "locked.log" until the year 2099 — Compliance
        // mode means even Governance bypass cannot delete it.
        let retain_until = chrono::DateTime::parse_from_rfc3339("2099-01-01T00:00:00Z")
            .expect("parse")
            .with_timezone(&Utc);
        lock_mgr.set(
            "b",
            "locked.log",
            ObjectLockState {
                mode: Some(LockMode::Compliance),
                retain_until: Some(retain_until),
                legal_hold_on: false,
            },
        );
        let s4 = Arc::new(
            S4Service::new(backend, registry, dispatcher)
                .with_lifecycle(Arc::clone(&mgr))
                .with_object_lock(Arc::clone(&lock_mgr)),
        );

        let report = run_scan_once(&s4).await.expect("scan");
        assert_eq!(report.buckets_scanned, 1);
        assert_eq!(report.objects_evaluated, 2);
        assert_eq!(report.expired, 1, "free.log should have been expired");
        assert_eq!(report.skipped_locked, 1, "locked.log must be skipped");
        assert_eq!(report.action_errors, 0);
    }

}