#!/bin/bash
# ==============================================================================================
# ============================= Rsync Core Script ==============================================
# ==============================================================================================
#
# PURPOSE
# ─────────────────────────────────────────────────────────────────────────────
# Core rsync engine for the two-server ecosystem. Called per share or per
# appdata profile by orchestrators (daily_sync_maintenance, weekly_sync_maintenance,
# critical_sync_maintenance) and directly for manual or scheduled dirty syncs.
#
# Profile is inferred from the directory basename (lowercased). Override with
# --profile=name for explicit selection. If no profile matches, global defaults
# from master.conf apply and no containers are stopped.
#
# After each sync, logs transfer data to bandwidth_monitor.sh for the weekly
# bandwidth report. Silent on success — only failures produce visible output.
#
# ==============================================================================================
# OPERATIONAL MODEL
# ==============================================================================================
#
# Profiles define per-share behavior:
#   PROFILE_CRITICAL_CONTAINER_NAMES  — containers stopped on both servers before sync
#   PROFILE_DELAYED_CONTAINERS        — containers with a delay before restart after sync
#   PROFILE_CONTAINER_DELAY           — seconds before delayed containers start
#   PROFILE_RSYNC_OPTS                — rsync flags (does not inherit DEFAULT_RSYNC_OPTS)
#   PROFILE_BW_LIMIT                  — bandwidth limit in KB/s
#   PROFILE_RETRY_COUNT               — retry attempts on failure
#   PROFILE_SLEEP                     — seconds between retry attempts
#   PROFILE_EXCLUDE_DIRS              — paths excluded from transfer
#   PROFILE_REMOTE_RESTART_CONTAINERS — containers restarted on remote after dirty sync
#                                       Was running → restart. Was stopped → leave stopped.
#
# Two-tier rsync enable/disable:
#   Tier 1: RSYNC_ENABLED=false → all rsync stops immediately (checked by this script)
#   Tier 2: per-orchestrator flag (DAILY_RSYNC_ENABLED etc.) → checked by caller
#
# Bandwidth logging: after each sync, logs profile/duration/status/bytes to
# bandwidth_monitor.sh --log-transfer. Bytes captured from rsync --stats via awk
# using version-stable field names.
#
# ==============================================================================================
# DESIGN PRINCIPLES
# ==============================================================================================
#
# Deletion Is Opt-In, Never Inherited
#   DEFAULT_RSYNC_OPTS deliberately omits --delete. Media shares only ever gain files
#   here; the arr cleanup scripts own deletion and are the only things that understand
#   whether a file is genuinely orphaned. --delete appears in exactly two places: a
#   profile that sets it explicitly, and the push pass of --merge-run.
#
# Profiles Replace, Not Extend
#   PROFILE_RSYNC_OPTS does not inherit DEFAULT_RSYNC_OPTS. A profile states its full
#   flag set, so reading one profile tells you exactly what will run — no tracing
#   through a base list to discover an inherited --delete.
#
# Local Is Authoritative
#   In --merge-run the remote may contribute content the local lacks, but never a
#   competing version. Pass 1 pulls only with --ignore-existing, so the push in pass 2
#   can safely be authoritative. Reversing that order would let the remote overwrite
#   local files before the delete pass.
#
# Pre-flight Before Payload
#   Connectivity, version parity, remote rootfs, remote share, remote disks and drive
#   temperature are all checked before a byte moves. A transfer aborted halfway is more
#   expensive to reason about than one that never started.
#
# Bounded, Resumable Transfers
#   No single attempt may exceed RSYNC_MAX_RUNTIME_HOURS. This is safe only because
#   --partial is in the default opts: a terminated transfer resumes rather than
#   restarting, so bounding it costs nothing and prevents one huge or stuck transfer
#   from holding its lock indefinitely and starving every other profile of a turn.
#
# Silent on Success
#   Only failures produce visible output. A quiet run is a successful one.
#
# ==============================================================================================
# OPERATIONAL SAFEGUARDS
# ==============================================================================================
#
# Root Enforcement
#   rsync over SSH as root and container stop/start both require it.
#
# Docker Presence Check
#   Verified before any profile container operations.
#
# Source Path Depth Guard
#   The source must be an absolute path at least three levels deep. It is pushed to
#   root@remote at the same absolute path and --merge-run adds --delete, so a truncated
#   argument is a remote-side hazard: /mnt/user would sync every share at once.
#
# Source Existence Check
#   The local directory must exist. Without this, --merge-run's pull pass would create
#   a mistyped directory, populate it from the remote, then push it back with --delete.
#
# Global Rsync Gate
#   check_rsync_enabled() — RSYNC_ENABLED=false exits cleanly before any operation.
#
# Partnership Blocklist
#   Refuses to sync if REMOTE_SERVER_NAME appears in the partnership blocklist.
#   Written at offboard — prevents stale access after a partnership ends.
#
# Version Parity
#   check_os_version_parity — refuses sync if servers on incompatible unRAID versions.
#
# Remote Health Pre-flights
#   check_connectivity() — Tailscale IP reachable before any SSH
#   check_remote_rootfs() — aborts if remote rootfs exceeds ROOTFS_WARN
#   check_remote_share() — aborts if target directory missing or empty on remote
#   check_remote_disks() — verifies all backing disks online on remote
#
# Drive Temperature Check
#   check_local_disk_temps() — runs before any transfer. Exit 1 = skip this profile,
#   exit 2 = abort all remaining syncs (CRITICAL temperature).
#
# Remote Docker Daemon Check
#   check_remote_docker_daemon — verified before any container stop/start operations.
#   If daemon unresponsive: container operations skipped, rsync proceeds without stopping.
#
# Per-Profile Concurrency Lock
#   acquire_rsync_lock() — per-profile lock prevents parallel runs of the same profile.
#   Global concurrent limit prevents too many simultaneous rsync processes.
#
# Runtime Ceiling
#   RSYNC_MAX_RUNTIME_HOURS terminates a single attempt that overruns, releasing the
#   per-profile lock for the next scheduled run. --partial in the default opts means the
#   paused transfer resumes rather than restarting from scratch.
#
# Merge-Run Ordering
#   Pass 1 pulls remote-unique content with --ignore-existing before pass 2 pushes with
#   --delete. Content the remote had and the local did not is preserved locally before
#   anything is deleted remotely.
#
# Merge-Run Delete Interlock
#   --delete is applied only when pass 1 completed. Its entire justification is that the
#   local is now the authoritative superset, and a capped or failed pull means it is not:
#   the remote still holds content the local never received. On an incomplete pull the
#   push proceeds without --delete and notifies, so local content still propagates while
#   nothing remote-unique is destroyed. The delete happens on a later run whose pull
#   succeeded.
#
# ==============================================================================================
# CONFIGURATION
# ==============================================================================================
#
# master.conf
#
#   RSYNC_ENABLED
#       Global on/off toggle for all rsync operations. (default: true)
#
#   DEFAULT_RSYNC_OPTS
#       Base rsync flags for unproiled shares. Note: --delete is intentionally absent —
#       media shares spread files only, arr cleanup scripts own deletions. Profile-specific
#       opts set --delete explicitly where needed.
#
#   BW_LIMIT
#       Default bandwidth cap in KB/s when no PROFILE_BW_LIMIT is set. (default: 0 = unlimited)
#
#   RETRY_COUNT
#       Default retry attempts on rsync failure. (default: 3)
#
#   SLEEP
#       Default seconds between retry attempts. (shipped default: 300)
#
#   RSYNC_MAX_RUNTIME_HOURS
#       Max hours a single transfer attempt may run before it's terminated and paused
#       for the next scheduled run. Protects the per-profile lock from being held
#       indefinitely by one huge/stuck transfer, starving other profiles of a turn.
#       Safe because DEFAULT_RSYNC_OPTS includes --partial — a paused transfer resumes
#       from where it left off, not from scratch. (code fallback 23; shipped conf sets 19)
#
#   ROOTFS_WARN
#       Abort threshold for remote rootfs percentage full. (default: 75)
#
#   PROFILES["profile_KEY"]
#       Profile definitions — one entry per PROFILE_* key per profile name.
#       See OPERATIONAL MODEL above for all supported keys.
#
#   BANDWIDTH_LOG / BANDWIDTH_WARN_GB
#       Shared with bandwidth_monitor.sh — set once, used by both.
#
# ==============================================================================================
# RUNTIME MODES
# ==============================================================================================
#
# rsync.sh /path/to/share
#     Sync the given path to the remote. Profile inferred from directory basename.
#
# rsync.sh /path/to/share --profile=name
#     Sync with explicit profile override — bypasses basename inference.
#
# rsync.sh /path/to/share --dry-run
#     Run all pre-flight checks and show what rsync would transfer. No transfer,
#     no container stops.
#
# rsync.sh /path/to/share --status
#     Show resolved profile, remote identity, and configuration. Then exit.
#
# rsync.sh /path/to/share --log
#     Verbose output throughout — every decision logged.
#
# rsync.sh /path/to/share --seed
#     Skip the empty-remote-share guard. Use for first-time seeding of a new share.
#
# rsync.sh /path/to/share --merge-run
#     Bidirectional merge: pull remote-unique content to local first (--ignore-existing),
#     then push local → remote with --delete so the remote matches local exactly.
#     Local is always authoritative — remote loses divergent file versions but keeps
#     any content the local did not have (pulled in pass 1 before the delete push).
#     Also auto-triggered when pre-scan detects ≥75% directory overlap with remote.
#
# ==============================================================================================

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

source "$SCRIPT_DIR/../load_config.sh"

# ── Separate positional directory arg from flags ───────────────────────────────────────────────
DIRECTORY=""
PROFILE_OVERRIDE=""
SEED=false
MERGE_RUN=false
RAW_ARGS=()

for ARG in "$@"; do
    case "$ARG" in
        --profile=*) PROFILE_OVERRIDE="${ARG#--profile=}" ;;
        --seed)      SEED=true ;;
        --merge-run) MERGE_RUN=true ;;
        --*|*=*)     RAW_ARGS+=("$ARG") ;;
        *)           [[ -z "$DIRECTORY" ]] && DIRECTORY="$ARG" || RAW_ARGS+=("$ARG") ;;
    esac
done

parse_args "${RAW_ARGS[@]}"

[[ -z "$DIRECTORY" ]] && {
    error "No directory specified"
    error "Usage: rsync.sh <dir> [--dry-run] [--log] [--profile=name]"
    exit 1
}

# ── Source path guards ────────────────────────────────────────────────────────────────────────
# This script pushes to root@remote at the same absolute path, and --merge-run adds --delete on
# the push pass. A truncated or mistyped source is therefore a remote-side data hazard, not just
# a local no-op: /mnt/user would sync every share at once, and / would target the filesystem
# root. Every real job path (see HOST*_*_SYNC_SHARES) is at least three levels deep.
_rsync_depth="${DIRECTORY//[^\/]/}"
if [[ "$DIRECTORY" != /* || "${#_rsync_depth}" -lt 3 ]]; then
    error "Refusing unsafe source path: '$DIRECTORY' — expected an absolute path at least 3 levels deep"
    exit 1
fi
unset _rsync_depth

# Without this, --merge-run's pull pass would create a mistyped local directory, fill it with
# remote content, then push it back with --delete.
if [[ ! -d "$DIRECTORY" ]]; then
    error "Source directory does not exist locally: $DIRECTORY"
    exit 1
fi

# ==============================================================================================
# ━━━ Setup ━━━
# ==============================================================================================
if [[ "$EUID" -ne 0 ]]; then
    error "Must be run as root"
    exit 1
fi

if ! command -v docker &>/dev/null; then
    error "Docker command not found"
    exit 1
fi


detect_hosts
require_partnership

# Tier 1 global gate — Tier 2 (per-orchestrator) checked by caller
if ! check_rsync_enabled; then
    warn "RSYNC_ENABLED=false — exiting cleanly"
    exit 0
fi

# Blocklist gate — refuse to sync with a partner blocked after offboard
BLOCKLIST_FILE="${PARTNERSHIP_BLOCKLIST_FILE:-${STATE_DIR}/partnership_blocklist.db}"
if [[ -f "$BLOCKLIST_FILE" ]] && grep -q "^${REMOTE_SERVER_NAME}|" "$BLOCKLIST_FILE" 2>/dev/null; then
    error "Rsync blocked — $REMOTE_SERVER_NAME is on the partnership blocklist"
    error "Re-onboard the partnership to restore access: partnership_manager.sh --onboard"
    exit 1
fi

resolve_remote_ip

# ── Profile inference ─────────────────────────────────────────────────────────────────────────
if [[ -n "$PROFILE_OVERRIDE" ]]; then
    PROFILE_NAME="$PROFILE_OVERRIDE"
    log "Profile override: $PROFILE_NAME"
else
    PROFILE_NAME=$(basename "$DIRECTORY" | tr '[:upper:]' '[:lower:]')
    log "Profile inferred: $PROFILE_NAME"
fi

# Acquire per-profile lock and check global concurrent limit
acquire_rsync_lock "$PROFILE_NAME"

# Tee all output to a live log file for the Varaverk UI
VV_LIVE_LOG="/tmp/unraid_locks/rsync_${PROFILE_NAME}.log"
VV_LAST_LOG="/tmp/unraid_locks/rsync_${PROFILE_NAME}.last.log"
: > "$VV_LIVE_LOG"
exec 1> >(tee -a "$VV_LIVE_LOG") 2>&1

# ── Load profile settings ─────────────────────────────────────────────────────────────────────
# NOTE: DEFAULT_RSYNC_OPTS and PROFILE_RSYNC_OPTS are bash arrays defined in master.conf —
# any "$BW_LIMIT" inside them is expanded once, when master.conf is sourced, before this
# override runs. So this recomputed $BW_LIMIT only affects the log line below and --status;
# it does NOT change the actual rsync --bwlimit unless a profile also bakes its own
# --bwlimit="${PROFILE_BW_LIMIT[name]}" directly into that profile's PROFILE_RSYNC_OPTS entry.
BW_LIMIT=${PROFILE_BW_LIMIT[$PROFILE_NAME]:-$BW_LIMIT}
RETRY_COUNT=${PROFILE_RETRY_COUNT[$PROFILE_NAME]:-$RETRY_COUNT}
SLEEP=${PROFILE_SLEEP[$PROFILE_NAME]:-$SLEEP}
CONTAINER_DELAY=${PROFILE_CONTAINER_DELAY[$PROFILE_NAME]:-$CONTAINER_DELAY}
RSYNC_MAX_RUNTIME_HOURS=${RSYNC_MAX_RUNTIME_HOURS:-23}
RSYNC_MAX_RUNTIME_SECONDS=$(( RSYNC_MAX_RUNTIME_HOURS * 3600 ))

read -r -a CRITICAL_CONTAINER_NAMES       <<< "${PROFILE_CRITICAL_CONTAINER_NAMES[$PROFILE_NAME]:-}"
read -r -a DELAYED_CONTAINERS             <<< "${PROFILE_DELAYED_CONTAINERS[$PROFILE_NAME]:-}"
read -r -a EXCLUDE_DIRS                   <<< "${PROFILE_EXCLUDE_DIRS[$PROFILE_NAME]:-}"
read -r -a REMOTE_RESTART_CONTAINERS      <<< "${PROFILE_REMOTE_RESTART_CONTAINERS[$PROFILE_NAME]:-}"

# Local containers use same names as remote (mirrored naming scheme)
LOCAL_CRITICAL_CONTAINER_NAMES=("${CRITICAL_CONTAINER_NAMES[@]}")

log "$ICON_GEAR Config: profile=${PROFILE_NAME} bw-limit=${BW_LIMIT}KB/s retry=${RETRY_COUNT} sleep=${SLEEP}s container-delay=${CONTAINER_DELAY}s"
log "$ICON_GEAR Containers: critical=${CRITICAL_CONTAINER_NAMES[*]:-none} delayed=${DELAYED_CONTAINERS[*]:-none} remote-restart=${REMOTE_RESTART_CONTAINERS[*]:-none}"

[[ "$SHOW_STATUS" == true ]] && show_status && exit 0
[[ "$DRY_RUN" == true ]] && warn "DRY RUN — no changes will be made"

# ==============================================================================================
# ━━━ Pre-flight Checks ━━━
# ==============================================================================================
echo ""
echo "━━━ $ICON_SHIELD Pre-flight ━━━"

# Disk temp — before touching remote or moving data
# Exit 1 = skip this profile | Exit 2 = abort all remaining profiles
check_local_disk_temps
TEMP_RESULT=$?
if [[ "$TEMP_RESULT" -eq 2 ]]; then
    error "Drive temps CRITICAL — aborting all remaining syncs"
    exit 2
elif [[ "$TEMP_RESULT" -eq 1 ]]; then
    warn "Drive temps high — skipping profile [$PROFILE_NAME]"
    exit 1
else
    log "Drive temps OK — $TEMP_CHECK_RESULT"
fi

# Version parity — refuse if servers on incompatible unRAID versions
check_os_version_parity || exit 1

check_connectivity
check_remote_rootfs
[[ "$SEED" == false ]] && check_remote_share "$DIRECTORY"
check_remote_disks "$DIRECTORY"

# ── Merge-run pre-scan ────────────────────────────────────────────────────────────────────────
# Auto-promote to merge mode when ≥75% of remote's top-level entries exist locally.
# Skipped when: --merge-run is already set, --seed is active, a named profile is resolved,
# or RSYNC_MERGE_ENABLED=false in master.conf.
if [[ "$MERGE_RUN" == false && "$SEED" == false \
    && "${RSYNC_MERGE_ENABLED:-true}" != "false" \
    && -z "${PROFILE_RSYNC_OPTS[$PROFILE_NAME]+x}" ]]; then
    _remote_top=$(ssh -i "$SSH_KEY" -o ConnectTimeout=10 -o BatchMode=yes \
        root@"$REMOTE_SERVER" "ls -1A '$DIRECTORY' 2>/dev/null | sort" 2>/dev/null)
    _remote_count=$(echo "$_remote_top" | grep -c .)
    _remote_count=${_remote_count:-0}

    if [[ "$_remote_count" -gt 0 ]]; then
        _local_top=$(ls -1A "$DIRECTORY" 2>/dev/null | sort)
        _overlap=$(comm -12 \
            <(echo "$_local_top") \
            <(echo "$_remote_top") | grep -c .)
        _overlap=${_overlap:-0}
        _overlap_pct=$(( _overlap * 100 / _remote_count ))

        if [[ "$_overlap_pct" -ge 75 ]]; then
            info "Overlap ${_overlap_pct}% (${_overlap}/${_remote_count} entries) — auto-promoting to merge mode"
            MERGE_RUN=true
        else
            log "Overlap ${_overlap_pct}% (${_overlap}/${_remote_count} entries) — normal sync"
        fi
    fi
    unset _remote_top _remote_count _local_top _overlap _overlap_pct
fi

# Remote Docker daemon — check before attempting container operations
if [[ ${#CRITICAL_CONTAINER_NAMES[@]} -gt 0 ]] || [[ ${#REMOTE_RESTART_CONTAINERS[@]} -gt 0 ]]; then
    check_remote_docker_daemon || {
        warn "Remote Docker daemon unresponsive — skipping container operations"
        warn "Proceeding with rsync only — containers will not be stopped or restarted"
        CRITICAL_CONTAINER_NAMES=()
        LOCAL_CRITICAL_CONTAINER_NAMES=()
        REMOTE_RESTART_CONTAINERS=()
    }
fi

# ==============================================================================================
# ━━━ Stop Containers ━━━
# ==============================================================================================
if [[ ${#CRITICAL_CONTAINER_NAMES[@]} -gt 0 || ${#LOCAL_CRITICAL_CONTAINER_NAMES[@]} -gt 0 ]]; then
    echo ""
    echo "━━━ $ICON_STOP $ICON_CONTAINERS Stop Containers ━━━"
    if [[ "$DRY_RUN" == true ]]; then
        for c in "${LOCAL_CRITICAL_CONTAINER_NAMES[@]:-}"; do
            [[ -n "$c" ]] && warn "DRY RUN — would stop local: $c"
        done
        for c in "${CRITICAL_CONTAINER_NAMES[@]:-}"; do
            [[ -n "$c" ]] && warn "DRY RUN — would stop remote: $c"
        done
    else
        # Local first — flush local databases before pushing
        stop_local_containers
        # Remote next — prevent writes while receiving
        stop_containers
    fi
fi

# ==============================================================================================
# ━━━ Transfer ━━━
# ==============================================================================================
echo ""
echo "━━━ $ICON_SYNC Transfer ━━━"
echo "$ICON_RUN  Source:   $DIRECTORY"
echo "$ICON_NET  Remote:   $REMOTE_SERVER:$DIRECTORY"
echo "$ICON_GEAR Profile:  $PROFILE_NAME"
echo "$ICON_HOST Identity: $MY_ID → $REMOTE_ID"
echo ""

get_rsync_opts

# Append profile excludes
for ex in "${EXCLUDE_DIRS[@]:-}"; do
    [[ -n "$ex" ]] && RSYNC_OPTS+=(--exclude="$ex")
done

# Add --stats to capture bytes transferred for bandwidth logging
RSYNC_OPTS+=(--stats)

# How the byte count is read back out of --stats. Both parse sites share this, because they had
# the same bug and fixing one would have left the other silently wrong.
#
# rsync prints:   Total transferred file size: 1,234,567 bytes
# so $NF is the word "bytes". The previous version stripped non-digits from $NF, which left an
# empty string, and "" + 0 is 0 — every transfer ever logged recorded 0 bytes. Nothing failed:
# bandwidth_history.db filled up correctly with 1,423 rows all reading zero, the graph plotted a
# flat line, and DIGEST_SMART_ON_BANDWIDTH could never fire because 0 is never above
# BANDWIDTH_WARN_GB. A whole feature that ran on schedule and measured nothing.
#
# Walks back from the end and takes the first field that is entirely digits once commas are
# removed, rather than trusting a fixed position: the units word is not guaranteed to be present
# on every rsync build, and $(NF-1) would then be as wrong as $NF is now.
VV_RSYNC_BYTES_AWK='
/Total transferred file size:/ {
    for (i = NF; i > 0; i--) {
        v = $i; gsub(/,/, "", v)
        if (v ~ /^[0-9]+$/) { print v + 0; exit }
    }
    print 0; exit
}'

[[ "$DRY_RUN" == true ]] && RSYNC_OPTS+=("--dry-run")

# ── Merge pass 1: pull remote-unique content to local (--ignore-existing) ────────────────────
# Runs before the push so anything the remote has that we don't is preserved locally.
# After this pass, local is the superset — the delete push in pass 2 is then safe.
if [[ "$MERGE_RUN" == true ]]; then
    echo "$ICON_SYNC  Merge pass 1 — pulling ${REMOTE_SERVER_NAME}-unique content to local (capped at ${RSYNC_MAX_RUNTIME_HOURS}h)..."
    _pull_opts=(-av --ignore-existing --stats)
    [[ "$DRY_RUN" == true ]] && _pull_opts+=(--dry-run)
    _pull_output=$(timeout "${RSYNC_MAX_RUNTIME_SECONDS}s" rsync "${_pull_opts[@]}" \
        -e "ssh -i $SSH_KEY -T -o Compression=no -o IPQoS=throughput" \
        "root@${REMOTE_SERVER}:${DIRECTORY}/" "${DIRECTORY}/" 2>&1)
    _pull_exit=$?
    _pull_bytes=$(echo "$_pull_output" | \
        awk "$VV_RSYNC_BYTES_AWK")
    _pull_complete=false
    if [[ "$_pull_exit" -eq 0 ]]; then
        echo "$ICON_DONE Merge pass 1 complete — ${_pull_bytes:-0} bytes pulled"
        _pull_complete=true
    elif [[ "$_pull_exit" -eq 124 ]]; then
        # Same cap as the main push — never let one direction of the merge run unbounded.
        # Files already completed before the timeout stay pulled; next scheduled run
        # picks up whatever's still remote-unique (no --partial here, so an in-flight
        # file at the moment of the kill is discarded, not corrupted).
        warn "Merge pass 1 exceeded ${RSYNC_MAX_RUNTIME_HOURS}h cap — pull incomplete"
    else
        warn "Merge pass 1 failed (exit $_pull_exit)"
        echo "$_pull_output" | grep -iE "error|rsync:|permission denied" | while read -r _line; do
            warn "  $_line"
        done
    fi

    # Pass 2: push with --delete — but ONLY if pass 1 actually finished.
    # --delete is justified solely by "local is now the authoritative superset", and that
    # premise holds only on a complete pull. If pass 1 was capped or failed, the remote still
    # holds content the local never received; deleting it here would destroy it permanently
    # and make the "resume the pull next run" promise impossible to keep. Push without
    # --delete instead — local content still propagates, nothing remote-unique is lost, and
    # the delete happens on a later run whose pull completed.
    if [[ "$_pull_complete" == true ]]; then
        RSYNC_OPTS+=(--delete)
    else
        warn "Skipping --delete this run — pass 1 did not complete, remote may hold content the local lacks"
        warn "Local content still pushes; the delete pass runs once a full pull succeeds"
        notify "Merge run on $(hostname) pushed without --delete — pass 1 incomplete for $PROFILE_NAME" \
            "Rsync" "warning"
    fi
    unset _pull_opts _pull_output _pull_exit _pull_bytes _pull_complete
fi

START=$(date +%s)
RSYNC_SUCCESS=false
RSYNC_TIMED_OUT=false
BYTES_TRANSFERRED=0
ATTEMPT=0

for (( ATTEMPT=1; ATTEMPT<=RETRY_COUNT; ATTEMPT++ )); do
    log "$ICON_RETRY Attempt $ATTEMPT of $RETRY_COUNT..."
    echo "$ICON_SYNC  Rsync running — this may take a while (capped at ${RSYNC_MAX_RUNTIME_HOURS}h)..."

    RSYNC_OUTPUT=$(timeout "${RSYNC_MAX_RUNTIME_SECONDS}s" rsync "${RSYNC_OPTS[@]}" \
        -e "ssh -i $SSH_KEY -T -o Compression=no -o IPQoS=throughput" \
        "$DIRECTORY" "root@${REMOTE_SERVER}:$(dirname "$DIRECTORY")/" 2>&1)

    RSYNC_EXIT=$?

    if [[ "$RSYNC_EXIT" -eq 0 ]]; then
        # Parse bytes transferred from --stats output
        BYTES_TRANSFERRED=$(echo "$RSYNC_OUTPUT" | \
            awk "$VV_RSYNC_BYTES_AWK")
        BYTES_TRANSFERRED="${BYTES_TRANSFERRED:-0}"

        echo "$ICON_DONE Rsync complete — $BYTES_TRANSFERRED bytes transferred"
        RSYNC_SUCCESS=true
        break
    elif [[ "$RSYNC_EXIT" -eq 124 ]]; then
        # Hit the runtime cap, not a failure — --partial means next run resumes from here.
        # No retry: retrying an already-huge transfer 2 more times just wastes the window.
        warn "$ICON_RETRY Rsync exceeded ${RSYNC_MAX_RUNTIME_HOURS}h cap — pausing, will resume next scheduled run"
        RSYNC_TIMED_OUT=true
        break
    else
        warn "$ICON_RETRY Rsync failed (attempt $ATTEMPT/$RETRY_COUNT)"
        log "Exit code: $RSYNC_EXIT"
        if [[ "$ATTEMPT" -lt "$RETRY_COUNT" ]]; then
            log "Retrying in ${SLEEP}s..."
            sleep "$SLEEP"
        fi
    fi
done

# ==============================================================================================
# ━━━ Start Containers ━━━
# ==============================================================================================
if [[ ${#CRITICAL_CONTAINER_NAMES[@]} -gt 0 || ${#LOCAL_CRITICAL_CONTAINER_NAMES[@]} -gt 0 ]]; then
    echo ""
    echo "━━━ $ICON_START $ICON_CONTAINERS Start Containers ━━━"
    if [[ "$DRY_RUN" == true ]]; then
        for c in "${CRITICAL_CONTAINER_NAMES[@]:-}"; do
            [[ -n "$c" ]] && warn "DRY RUN — would start remote: $c"
        done
        for c in "${LOCAL_CRITICAL_CONTAINER_NAMES[@]:-}"; do
            [[ -n "$c" ]] && warn "DRY RUN — would start local: $c"
        done
    else
        # Remote first — can be coming up while local restarts
        start_containers
        # Local next
        start_local_containers
    fi
fi

# ==============================================================================================
# ━━━ Remote Restart (dirty sync profiles) ━━━
# ==============================================================================================
# For dirty sync profiles (critical-fallback) — restart containers on remote
# that were running before sync so they pick up config changes from the dirty sync window.
# Was running → restart. Was stopped → leave stopped.
if [[ ${#REMOTE_RESTART_CONTAINERS[@]} -gt 0 ]]; then
    echo ""
    echo "━━━ $ICON_START $ICON_CONTAINERS Remote Restart (post dirty sync) ━━━"
    log "Restarting configured containers on $REMOTE_SERVER_NAME..."

    for container in "${REMOTE_RESTART_CONTAINERS[@]}"; do
        [[ -z "$container" ]] && continue

        # Check if container was running before sync (still tracked via RUNNING_CONTAINERS)
        WAS_RUNNING=false
        for prev in "${RUNNING_CONTAINERS[@]:-}"; do
            [[ "$prev" == "$container" ]] && WAS_RUNNING=true && break
        done

        if [[ "$WAS_RUNNING" == false ]]; then
            # Not in stop list — check current remote state
            REMOTE_STATUS=$(timeout 15 ssh -i "$SSH_KEY" \
                -o ConnectTimeout=10 root@"$REMOTE_SERVER" \
                "docker inspect -f '{{.State.Running}}' $container 2>/dev/null" 2>/dev/null)
            [[ "$REMOTE_STATUS" != "true" ]] && \
                log "$container not running on $REMOTE_SERVER_NAME — skipping remote restart" && \
                continue
        fi

        if [[ "$DRY_RUN" == true ]]; then
            warn "DRY RUN — would restart $container on $REMOTE_SERVER_NAME"
            continue
        fi

        timeout 15 ssh -i "$SSH_KEY" -o ConnectTimeout=10 root@"$REMOTE_SERVER" \
            "docker restart $container" >/dev/null 2>&1 && \
            echo "$ICON_STARTED $container restarted on $REMOTE_SERVER_NAME ✅" || \
            warn "Failed to restart $container on $REMOTE_SERVER_NAME"
    done
fi

END=$(date +%s)
DURATION=$(( END - START ))

# ==============================================================================================
# ━━━ Bandwidth Logging ━━━
# ==============================================================================================
# Logs to bandwidth_monitor.sh — new format includes bytes transferred and warn flag.
# Only logs on actual runs (not dry-run) and only when bandwidth_monitor.sh exists.
BANDWIDTH_MONITOR="$SCRIPT_DIR/../Monitors/bandwidth_monitor.sh"

if [[ "$DRY_RUN" == false ]] && [[ -f "$BANDWIDTH_MONITOR" ]]; then
    STATUS="success"
    [[ "$RSYNC_SUCCESS" == false ]] && STATUS="failed"
    [[ "$RSYNC_TIMED_OUT" == true ]] && STATUS="timeout"
    bash "$BANDWIDTH_MONITOR" --log-transfer \
        "$PROFILE_NAME" "$DURATION" "$STATUS" "$BYTES_TRANSFERRED"
    log "$ICON_BANDWIDTH Transfer logged to bandwidth monitor ($BYTES_TRANSFERRED bytes)"
fi

# ==============================================================================================
# ━━━ Summary ━━━
# ==============================================================================================
echo ""
echo "━━━━━ $ICON_SUMMARY RSYNC SUMMARY ━━━━━"
echo "$ICON_HOST    Identity:  $MY_ID ($LOCAL_SERVER_NAME)"
echo "$ICON_RUN     Directory: $DIRECTORY"
echo "$ICON_GEAR    Profile:   $PROFILE_NAME"
echo "$ICON_TIME    Duration:  $(format_duration $DURATION)"
[[ "$BYTES_TRANSFERRED" -gt 0 ]] && \
    echo "$ICON_BANDWIDTH Transferred: $(awk "BEGIN {printf \"%.2fGB\", $BYTES_TRANSFERRED / 1073741824}")"

if [[ "$DRY_RUN" == true ]]; then
    warn "DRY RUN — no changes made"
elif [[ "$RSYNC_SUCCESS" == true ]]; then
    echo "$ICON_DONE Status: $ICON_SUCCESS DONE"
elif [[ "$RSYNC_TIMED_OUT" == true ]]; then
    echo "$ICON_DONE Status: PAUSED — exceeded ${RSYNC_MAX_RUNTIME_HOURS}h cap, resumes next scheduled run"
else
    echo "$ICON_ERROR Status: FAILED after $RETRY_COUNT attempts"
    notify "Rsync FAILED — $DIRECTORY ($PROFILE_NAME) after $RETRY_COUNT attempts on $(hostname)" \
        "Rsync" "warning"
fi
echo "━━━━━━━━━━━━━━━━━━━━━━━"

# Flush tee and preserve log for UI — close both ends of the pipe so tee gets EOF
exec 1>&- 2>&-; wait
cp "$VV_LIVE_LOG" "$VV_LAST_LOG" 2>/dev/null
rm -f "$VV_LIVE_LOG"

[[ "$RSYNC_SUCCESS" == false ]] && [[ "$RSYNC_TIMED_OUT" == false ]] && [[ "$DRY_RUN" == false ]] && exit 1
exit 0