#!/bin/bash
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
set -euo pipefail
set -o errtrace
trap 'echo "WARNING: error on line $LINENO: \"$BASH_COMMAND\""; exit 1' ERR
shopt -s nullglob
IFS=$'\n\t'
echo "Restic Backup V 4.5.4"

# Check if Already Running
exec 9>/var/run/smartstore.lock
flock -n 9 || {
  echo "WARNING: [$(date)] : process is already running"
  exit 1
}

if ! command -v restic &> /dev/null; then
    echo "restic not found. Starting installation..."
    apt-get update && apt-get install -y restic
fi

# Check dependencies early (before mounting and starting the backup)
for cmd in qm pct restic pvesm pvesh jq; do
    command -v "$cmd" &>/dev/null || { echo "WARNING: required command missing: ${cmd}"; exit 1; }
done


# Set Default Values
RESTIC_BASEFOLDER="/media/SmartStore"
logPath="/var/log/smartstore.log"
mountPersistent="no"
setStandby="yes"
fsTrim="yes"
UUIDS=()
MOUNTOPT=()
COUNTIGNORE="no"

# ---- ResticBackup defaults (can be overridden in config.cfg) ----
KEEP_DAILY="${KEEP_DAILY:-10}"
KEEP_WEEKLY="${KEEP_WEEKLY:-4}"
KEEP_MONTHLY="${KEEP_MONTHLY:-3}"
PRUNE_AFTER_BACKUP="${PRUNE_AFTER_BACKUP:-true}"
EXCLUDE_VMIDS="${EXCLUDE_VMIDS:-}"
SKIP_TEMPLATES="${SKIP_TEMPLATES:-true}"
NOTIFY_EMAIL="${NOTIFY_EMAIL:-root}"
NOTIFY_ON_SUCCESS="${NOTIFY_ON_SUCCESS:-true}"
LXC_EXCLUDE_DIRS="${LXC_EXCLUDE_DIRS:-/var/cache,/var/log}"
# Print the overview of all backups at the end of the run (one
# 'restic stats' call per snapshot).
PRINT_BACKUP_OVERVIEW="${PRINT_BACKUP_OVERVIEW:-true}"
RESTIC_COMPRESSION_SEND="${RESTIC_COMPRESSION_SEND:-off}"
# ---- Backup mode ----
# raw = a RAW image of every volume (the default and the right choice whenever
#       space on the target drive matters)
# zfs = zfs send for every machine with zvol backed disks
BACKUP_MODE="${BACKUP_MODE:-raw}"
# Individual DISKS that use zfs send although BACKUP_MODE=raw, as a comma
# separated list of "<vmid>:<disk>" descriptors, e.g. "100:scsi1,101:virtio2".
# Per disk on purpose: the big data disk of a machine can be worth it while its
# OS disk is not. Only zvol backed disks can use it; everything else is stored
# as a RAW image.
ZFS_SEND_DISKS="${ZFS_SEND_DISKS:-}"
# Point out disks that are large and barely change - the one case where zfs
# send still pays off, because it reads only the changed blocks.
ZFS_SEND_HINT="${ZFS_SEND_HINT:-true}"
ZFS_SEND_HINT_MIN_GB="${ZFS_SEND_HINT_MIN_GB:-500}"
ZFS_SEND_HINT_MAX_CHURN_PERCENT="${ZFS_SEND_HINT_MAX_CHURN_PERCENT:-1}"
RESTIC_COMPRESSION_RAW="${RESTIC_COMPRESSION_RAW:-auto}"
# Compression for all file based backups (LXC, LOCAL_DIRS, config).
RESTIC_COMPRESSION_FILES="${RESTIC_COMPRESSION_FILES:-auto}"
RAW_BLOCK_SIZE="${RAW_BLOCK_SIZE:-4M}"
RAW_USE_LVM_SNAPSHOT="${RAW_USE_LVM_SNAPSHOT:-true}"
# Without a consistent snapshot source do NOT write a RAW backup (no live read).
RAW_REQUIRE_SNAPSHOT="${RAW_REQUIRE_SNAPSHOT:-true}"
RAW_LVM_SNAPSHOT_SIZE="${RAW_LVM_SNAPSHOT_SIZE:-5G}"
FULL_MAX_AGE_DAYS="${FULL_MAX_AGE_DAYS:-21}"
LOCAL_DIRS="${LOCAL_DIRS:-}"
LOCAL_EXCLUDE_DIRS="${LOCAL_EXCLUDE_DIRS:-.zfs}"
RESTIC_CHECK_AFTER_BACKUP="${RESTIC_CHECK_AFTER_BACKUP:-true}"
RESTIC_CHECK_MAX_GB="${RESTIC_CHECK_MAX_GB:-10}"
# Walk the whole repository over several runs instead of sampling at random:
# run N reads group N of RESTIC_CHECK_ROTATE_RUNS, so after that many runs every
# pack file has been read back and verified once.
RESTIC_CHECK_ROTATE="${RESTIC_CHECK_ROTATE:-true}"
RESTIC_CHECK_ROTATE_RUNS="${RESTIC_CHECK_ROTATE_RUNS:-30}"
DISK_SPACE_WARN_PERCENT="${DISK_SPACE_WARN_PERCENT:-10}"
FULL_FILESYSTEM_PERCENTAGE="${FULL_FILESYSTEM_PERCENTAGE:-90}"
MAX_UPTIME_DAYS="${MAX_UPTIME_DAYS:-39}"
# Expected growth of an already existing file based / RAW backup,
# in percent of the space it currently occupies.
ESTIMATED_BACKUP_INCREASE_PERCENTAGE="${ESTIMATED_BACKUP_INCREASE_PERCENTAGE:-10}"
# On a total lack of space (not even freeing all old backups is enough) a
# prominent warning is printed when this is TRUE.
LOW_MEMORY_WARNING="${LOW_MEMORY_WARNING:-TRUE}"
RESTIC_DRY_RUN=false
RESTIC_VERBOSE=false

# Import config file (optional - the script also runs on defaults without it)
config=/usr/local/bin/config.cfg
if [[ -r "$config" ]]; then
    # shellcheck source=/dev/null
    source "$config"
else
    echo "INFO: no config found (${config}), using defaults"
fi


# Normalize the backup mode. BACKUP_MODE=zfs backs everything up in the classic
# zfs send format; in raw mode zfs send is opt-in per disk via ZFS_SEND_DISKS.
BACKUP_MODE="$(echo "${BACKUP_MODE}" | tr '[:upper:]' '[:lower:]')"
case "$BACKUP_MODE" in
    zfs|raw) : ;;
    *) echo "WARNING: invalid BACKUP_MODE '${BACKUP_MODE}' - using 'raw'"; BACKUP_MODE="raw" ;;
esac

# A malformed descriptor would never match anything and would silently keep
# costing a full read on every run, so the form is checked here.
ZFS_SEND_DISKS="${ZFS_SEND_DISKS//[[:space:]]/}"
if [[ -n "$ZFS_SEND_DISKS" ]]; then
    _zsd_bad=""
    _zsd_ok=""
    IFS=',' read -ra _zsd_list <<< "$ZFS_SEND_DISKS"
    for _zsd_e in "${_zsd_list[@]}"; do
        [[ -z "$_zsd_e" ]] && continue
        if [[ "$_zsd_e" =~ ^[0-9]+:[a-z]+[0-9]+$ ]]; then
            _zsd_ok+="${_zsd_ok:+,}${_zsd_e}"
        else
            _zsd_bad+=" '${_zsd_e}'"
        fi
    done
    if [[ -n "$_zsd_bad" ]]; then
        echo "WARNING: ZFS_SEND_DISKS expects <vmid>:<disk> entries like 100:scsi1 -"
        echo "         ignoring:${_zsd_bad}"
    fi
    ZFS_SEND_DISKS="$_zsd_ok"
    unset _zsd_bad _zsd_ok _zsd_list _zsd_e
fi

# restic groups the pack files by the FIRST BYTE of the pack ID, so it can tell
# at most 256 groups apart - a larger value would silently check nothing.
if ! [[ "$RESTIC_CHECK_ROTATE_RUNS" =~ ^[0-9]+$ ]] || [[ "$RESTIC_CHECK_ROTATE_RUNS" -lt 1 ]]; then
    echo "WARNING: invalid RESTIC_CHECK_ROTATE_RUNS '${RESTIC_CHECK_ROTATE_RUNS}' - using 30"
    RESTIC_CHECK_ROTATE_RUNS=30
elif [[ "$RESTIC_CHECK_ROTATE_RUNS" -gt 256 ]]; then
    echo "WARNING: RESTIC_CHECK_ROTATE_RUNS is capped at 256 by restic - using 256"
    RESTIC_CHECK_ROTATE_RUNS=256
fi

if ! command -v zfs &>/dev/null; then
    if [[ "$BACKUP_MODE" == "zfs" || -n "$ZFS_SEND_DISKS" ]]; then
        echo "WARNING: zfs send requested but zfs is missing on this host"
    fi
    echo "INFO: no ZFS on this host - all volumes are backed up as RAW images"
    BACKUP_MODE="raw"
    ZFS_SEND_DISKS=""
fi

# Derive repo and password path from RESTIC_BASEFOLDER
# (both can be overridden explicitly in the config)
RESTIC_REPOSITORY="${RESTIC_REPOSITORY:-${RESTIC_BASEFOLDER}/ResticBackup}"
RESTIC_PASSWORDFILE="${RESTIC_PASSWORDFILE:-${RESTIC_BASEFOLDER}/ResticBackup.pw}"

# ================== ResticBackup Runtime State ==================
ERRORS=0
WARNINGS=0
TOTAL_VMS=0
TOTAL_CTS=0
TOTAL_DISKS=0
TOTAL_SKIPPED=0
LOG_OUTPUT=""
FULL_RESTIC_ID=""
BASE_SNAP_NAME=""
# Name tag of the VM/CT currently being backed up (for restic --tag "name=...")
NAME_TAG_FLAGS=()
# Backup run identifier: short_id of this run's config snapshot.
# Appended to every disk backup as tag "config-id=<id>" so that a restore
# can find all disks that belong together.
CURRENT_CONFIG_ID=""
CONFIG_ID_TAG_FLAGS=()

# Harden tag values: commas would split the tag on the restic CLI,
# whitespace makes the evaluation fragile
sanitize_tag_value() {
    local s="$1"
    s="${s//,/}"
    s="${s//[[:space:]]/_}"
    echo "$s"
}
FULL_TIME=""
# Set once per run, as soon as the space check has been done.
FULL_SPACE_CHECK_DONE=false
# true when specific VMIDs were given on the command line, so only part of the
# host is in this run.
RB_PARTIAL_RUN=false
TOTAL_FULLS=0
TOTAL_DIFFS=0
TOTAL_RAWS=0
# Result of the planning phase: exact space required by this run, in bytes.
SUM_ZFS_BYTES=0      # zfs send streams (full or differential, per disk)
SUM_RAW_BYTES=0      # requirement of all RAW images
SUM_OTHER_BYTES=0    # requirement of all LXC and LOCAL_DIRS backups
REQUIRED_BYTES=0     # the space actually needed
VOL_PATH=""
VOL_KIND=""
RAW_READ_PATH=""
RAW_TMP_SNAP=""
RAW_TMP_CLONE=""
RAW_SOURCE_CONSISTENT=false
RAW_SNAPSHOT_ATTEMPTED=false
TOTAL_DATA_ADDED=0
REPO_SIZE_BEFORE=0

# ================== Logging ==================
# Every message goes to stdout AND into LOG_OUTPUT, which is mailed at the end
# of the run. log_warn/log_error additionally maintain the counters that decide
# the exit code and the subject line of the notification mail.

rb_log() {
    local level="$1"; shift
    local msg="[$(date '+%Y-%m-%dT%H:%M:%S%z')] [$level] $*"
    echo "$msg"
    LOG_OUTPUT+="${msg}"$'\n'
}
log_info()  { rb_log "INFO" "$@"; }
log_warn()  { rb_log "WARNING" "$@"; WARNINGS=$((WARNINGS+1)); }
log_error() { rb_log "ERROR" "$@"; ERRORS=$((ERRORS+1)); }

is_excluded() {
    local vmid="$1"
    [[ ",${EXCLUDE_VMIDS}," == *",${vmid},"* ]]
}

# ================== RAW image mode ==================

# Resolve a Proxmox volume and classify it.
# Sets VOL_PATH and VOL_KIND (zvol|blockdev|dir|file).
resolve_volume() {
    local volume="$1"
    VOL_PATH=""; VOL_KIND=""
    [[ -z "$volume" || "$volume" == "none" ]] && return 1
    if [[ "$volume" == /* ]]; then
        VOL_PATH="$volume"
    else
        VOL_PATH=$(pvesm path "$volume" 2>/dev/null) || return 1
    fi
    [[ -z "$VOL_PATH" ]] && return 1
    if [[ "$VOL_PATH" == /dev/zvol/* ]]; then
        VOL_KIND="zvol"
    elif [[ -b "$VOL_PATH" ]]; then
        VOL_KIND="blockdev"
    elif [[ -d "$VOL_PATH" ]]; then
        VOL_KIND="dir"
    elif [[ -f "$VOL_PATH" ]]; then
        VOL_KIND="file"
    else
        return 1
    fi
    return 0
}

# restic knows exactly three values. Anything else is caught at startup instead
# of failing on every backup call. $1 is the NAME of the setting to normalize.
normalize_compression() {
    local -n _c="$1"
    case "${_c,,}" in
        off|auto|max) _c="${_c,,}" ;;
        *)
            log_warn "${1}: \"${_c}\" is not a valid value (off|auto|max) - using auto"
            _c="auto"
            ;;
    esac
}

# Size of a RAW source in bytes (block device or file).
raw_source_size() {
    local p="$1" v=""
    if [[ -b "$p" ]]; then
        v=$(blockdev --getsize64 "$p" 2>/dev/null) || v=""
    fi
    [[ -z "$v" ]] && { v=$(stat -Lc %s "$p" 2>/dev/null) || v=""; }
    [[ "$v" =~ ^[0-9]+$ ]] || return 1
    echo "$v"
}

# Space actually occupied by a RAW source in bytes, holes excluded.
# Far more realistic for SPACE PLANNING than the logical size: unallocated
# ranges read back as zeros and end up in the repo deduplicated to nearly
# nothing. (The amount of data READ is still the full size - that is I/O
# effort, not a space requirement.)
#   ZFS zvol       : logicalreferenced of the newest snapshot, else the dataset
#   LVM-thin       : lv_size * data_percent
#   raw file       : allocated blocks (st_blocks), not the apparent size
#   otherwise      : logical size (thick provisioning has no sparseness)
raw_allocated_bytes() {
    local src="$1" v=""

    if [[ "$src" == /dev/zvol/* ]] && command -v zfs &>/dev/null; then
        local ds="${src#/dev/zvol/}" target=""
        if find_latest_snap "$ds"; then
            target="$LATEST_SNAP"
        else
            target="$ds"
        fi
        v=$(zfs get -Hp -o value logicalreferenced "$target" 2>/dev/null) || v=""
        if [[ ! "$v" =~ ^[0-9]+$ ]]; then
            v=$(zfs get -Hp -o value referenced "$target" 2>/dev/null) || v=""
        fi
        [[ "$v" =~ ^[0-9]+$ ]] && { echo "$v"; return 0; }
    fi

    if [[ -b "$src" ]] && command -v lvs &>/dev/null; then
        local lsize dpct
        lsize=$(lvs --noheadings --units b --nosuffix -o lv_size "$src" 2>/dev/null | tr -d '[:space:]') || lsize=""
        dpct=$(lvs --noheadings -o data_percent "$src" 2>/dev/null | tr -d '[:space:]') || dpct=""
        if [[ "$lsize" =~ ^[0-9]+$ && -n "$dpct" ]]; then
            v=$(awk -v s="$lsize" -v pc="$dpct" 'BEGIN{printf "%d", s*pc/100}') || v=""
            [[ "$v" =~ ^[0-9]+$ ]] && { echo "$v"; return 0; }
        fi
    fi

    if [[ -f "$src" ]]; then
        local blocks
        blocks=$(stat -Lc %b "$src" 2>/dev/null) || blocks=""
        [[ "$blocks" =~ ^[0-9]+$ ]] && { echo $(( blocks * 512 )); return 0; }
    fi

    raw_source_size "$src"
}

# Prepare a consistent read source:
#   LVM / LVM-thin -> temporary snapshot LV
#   ZFS zvol       -> snapshot device, if visible
#   otherwise      -> read directly (with a warning)
# Sets RAW_READ_PATH and, where applicable, RAW_TMP_SNAP.
raw_prepare_source() {
    local src="$1"
    RAW_READ_PATH="$src"
    RAW_TMP_SNAP=""
    RAW_TMP_CLONE=""
    RAW_SOURCE_CONSISTENT=false
    RAW_SNAPSHOT_ATTEMPTED=false

    # ZFS zvol: the snapshot EXISTS (find_latest_snap finds it), but it only
    # appears under /dev/zvol/ when snapdev=visible is set - the default is
    # hidden. So: take the device if it is there, otherwise make the snapshot
    # visible through a temporary clone (costs no space, COW).
    if [[ "$src" == /dev/zvol/* ]] && command -v zfs &>/dev/null; then
        local ds="${src#/dev/zvol/}"
        RAW_SNAPSHOT_ATTEMPTED=true
        if ! find_latest_snap "$ds"; then
            log_error "  RAW source: no ZFS snapshot exists on ${ds}"
            return 0
        fi

        local snapdev="/dev/zvol/${LATEST_SNAP}"
        if [[ -b "$snapdev" ]]; then
            RAW_READ_PATH="$snapdev"
            RAW_SOURCE_CONSISTENT=true
            log_info "  RAW source: ZFS snapshot ${LATEST_SNAP} (snapdev visible)"
            return 0
        fi

        # snapdev=hidden -> temporaeren Klon anlegen
        local clone="${ds}_rbclone" _orig=""
        _orig=$(zfs get -H -o value origin "$clone" 2>/dev/null) || _orig=""
        if [[ -n "$_orig" && "$_orig" != "-" ]]; then
            # Leftover from an earlier run - only real clones are removed
            zfs destroy -f "$clone" &>/dev/null || true
        fi
        if zfs clone -o volmode=dev "$LATEST_SNAP" "$clone" &>/dev/null \
           || zfs clone "$LATEST_SNAP" "$clone" &>/dev/null; then
            RAW_TMP_CLONE="$clone"
            local clonedev="/dev/zvol/${clone}" _i=0
            command -v udevadm &>/dev/null && udevadm settle --timeout=10 &>/dev/null || true
            while [[ ! -b "$clonedev" && $_i -lt 50 ]]; do sleep 0.1; _i=$((_i+1)); done
            if [[ -b "$clonedev" ]]; then
                RAW_READ_PATH="$clonedev"
                RAW_SOURCE_CONSISTENT=true
                log_info "  RAW source: ZFS snapshot ${LATEST_SNAP} via temporary clone ${clone}"
                return 0
            fi
            log_warn "  RAW source: clone device ${clonedev} did not appear"
            raw_cleanup_source
        else
            log_warn "  RAW source: 'zfs clone ${LATEST_SNAP} ${clone}' failed"
        fi
        log_warn "  RAW source: no consistent source for ${ds} (hint: zfs set snapdev=visible ${ds})"
        return 0
    fi

    [[ "$RAW_USE_LVM_SNAPSHOT" == true ]] || return 0
    [[ -b "$src" ]] || return 0
    command -v lvs &>/dev/null || return 0
    command -v lvcreate &>/dev/null || return 0

    RAW_SNAPSHOT_ATTEMPTED=true
    local vg lv pool
    vg=$(lvs --noheadings -o vg_name "$src" 2>/dev/null | tr -d '[:space:]') || vg=""
    lv=$(lvs --noheadings -o lv_name "$src" 2>/dev/null | tr -d '[:space:]') || lv=""
    if [[ -z "$vg" || -z "$lv" ]]; then
        RAW_SNAPSHOT_ATTEMPTED=false
        log_warn "  RAW source: not an LVM volume (${src}) - reading live"
        return 0
    fi
    pool=$(lvs --noheadings -o pool_lv "$src" 2>/dev/null | tr -d '[:space:]') || pool=""

    local snap="${lv}_rbsnap"
    lvremove -f "${vg}/${snap}" &>/dev/null || true
    if [[ -n "$pool" ]]; then
        lvcreate -s -n "$snap" "${vg}/${lv}" &>/dev/null || {
            log_warn "  RAW source: thin snapshot ${vg}/${snap} failed - reading live"; return 0; }
        lvchange -ay -Ky "${vg}/${snap}" &>/dev/null || true
    else
        lvcreate -s -L "$RAW_LVM_SNAPSHOT_SIZE" -n "$snap" "${vg}/${lv}" &>/dev/null || {
            log_warn "  RAW source: LVM snapshot ${vg}/${snap} failed - reading live"; return 0; }
    fi
    RAW_TMP_SNAP="${vg}/${snap}"
    RAW_READ_PATH="/dev/${vg}/${snap}"
    RAW_SOURCE_CONSISTENT=true
    log_info "  RAW source: LVM snapshot ${RAW_TMP_SNAP}"
}

raw_cleanup_source() {
    if [[ -n "$RAW_TMP_CLONE" ]]; then
        zfs destroy -f "$RAW_TMP_CLONE" &>/dev/null \
            || log_warn "  Could not remove temporary ZFS clone ${RAW_TMP_CLONE} (manually: zfs destroy ${RAW_TMP_CLONE})"
        RAW_TMP_CLONE=""
    fi
    [[ -z "$RAW_TMP_SNAP" ]] && return 0
    lvremove -f "$RAW_TMP_SNAP" &>/dev/null \
        || log_warn "  Could not remove LVM snapshot ${RAW_TMP_SNAP} (manually: lvremove -f ${RAW_TMP_SNAP})"
    RAW_TMP_SNAP=""
}

# If the run is aborted (Ctrl+C, SIGTERM, or the ERR trap above) the temporary
# artefacts of the current volume would stay behind: an LVM snapshot LV, a ZFS
# clone and the read-only snapshot mount used for file based LXC backups. They
# cost space and block the next run, so they are cleaned up unconditionally on
# exit. Everything is best effort - a failing cleanup must not mask the
# original exit code.
cleanup_on_exit() {
    local _rc=$?
    trap - EXIT INT TERM ERR
    raw_cleanup_source 2>/dev/null || true
    local _stale
    for _stale in /tmp/restic-snap-*; do
        [[ -d "$_stale" ]] || continue
        if mountpoint -q "$_stale" 2>/dev/null; then
            umount "$_stale" 2>/dev/null || true
        fi
        rmdir "$_stale" 2>/dev/null || true
    done
    exit "$_rc"
}
trap cleanup_on_exit EXIT INT TERM

# Write a RAW image of a volume to restic.
# Always a full backup - there are no differentials in this mode.
# $1 prefix (vm|ct)  $2 id  $3 disk_key  $4 Quellpfad
backup_raw_volume() {
    local prefix="$1" id="$2" disk_key="$3" src_path="$4"
    local disk_tag="${prefix}-${id}-${disk_key}"
    local filename="${prefix}-${id}-${disk_key}.raw"
    local size comp
    size=$(raw_source_size "$src_path") || size=0
    [[ "$size" =~ ^[0-9]+$ ]] || size=0
    comp="$RESTIC_COMPRESSION_RAW"

    if $RESTIC_DRY_RUN; then
        log_info "  [DRY-RUN] RAW image ${disk_key}: dd if=${src_path} | restic --stdin ${filename} (compression=${comp}, $(human_bytes "$size"))"
        return 0
    fi

    raw_prepare_source "$src_path"
    if [[ "$RAW_REQUIRE_SNAPSHOT" == true ]] && $RAW_SNAPSHOT_ATTEMPTED && ! $RAW_SOURCE_CONSISTENT; then
        raw_cleanup_source
        log_error "  ${disk_key}: no consistent snapshot source - skipped (RAW_REQUIRE_SNAPSHOT=true)"
        return 1
    fi
    log_info "  RAW image ${disk_key}: ${RAW_READ_PATH} ($(human_bytes "$size"), compression=${comp})"

    local _err
    _err=$(mktemp) || _err=""
    set +e +o pipefail
    dd if="$RAW_READ_PATH" bs="$RAW_BLOCK_SIZE" status=none 2>/dev/null | restic backup \
        --stdin --stdin-filename "$filename" --compression "$comp" \
        --tag "${prefix}-${id}" --tag "$disk_tag" --tag "raw" \
        --tag "raw-size=${size}" --tag "raw-source=${VOL_KIND:-unknown}" \
        "${NAME_TAG_FLAGS[@]}" "${CONFIG_ID_TAG_FLAGS[@]}" &>"${_err:-/dev/null}"
    local _ps=("${PIPESTATUS[@]}")
    set -e -o pipefail
    if [[ -n "$_err" ]]; then
        log_restic_output "raw ${disk_key}" "$(cat "$_err" 2>/dev/null)"
        rm -f "$_err"
    fi
    raw_cleanup_source

    local _dd_rc=${_ps[0]:-1}
    local _restic_rc=${_ps[1]:-1}
    if [[ "$_restic_rc" -eq 0 && "$_dd_rc" -eq 0 ]]; then
        log_info "  OK: ${disk_key} raw"
        TOTAL_DISKS=$((TOTAL_DISKS+1))
        TOTAL_RAWS=$((TOTAL_RAWS+1))
    elif [[ "$_restic_rc" -eq 0 ]]; then
        log_warn "  Partial: ${disk_key} raw (dd exit=${_dd_rc}) - removing incomplete backup"
        local _bad_snap
        _bad_snap=$(restic snapshots --tag "${disk_tag},raw" --json 2>/dev/null \
            | jq -r 'sort_by(.time) | last | .short_id // empty' 2>/dev/null) || true
        [[ -n "$_bad_snap" ]] && { restic forget "$_bad_snap" &>/dev/null || log_warn "  Could not remove ${_bad_snap}"; }
    else
        log_error "  Failed: ${filename} (restic=${_restic_rc}, dd=${_dd_rc})"
    fi
}

find_latest_snap() {
    local dataset="$1"
    LATEST_SNAP=""
    local snap
    snap=$(zfs list -t snapshot -H -o name -S creation "${dataset}" 2>/dev/null | head -1) || true
    if [[ -n "$snap" ]]; then
        LATEST_SNAP="$snap"
        return 0
    fi
    return 1
}

is_snap_stale() {
    local snap_full="$1"
    local creation_epoch now_epoch age
    # -p returns the creation time directly as an epoch (locale independent)
    creation_epoch=$(zfs get -Hp -o value creation "${snap_full}" 2>/dev/null) || return 1
    [[ "$creation_epoch" =~ ^[0-9]+$ ]] || return 1
    now_epoch=$(date +%s)
    age=$(( now_epoch - creation_epoch ))
    [[ $age -gt 86400 ]]
}

prepare_snapshot() {
    local dataset="$1"
    SNAP_FULL=""
    if ! find_latest_snap "$dataset"; then
        log_error "  No snapshot found on ${dataset}"
        return 1
    fi
    SNAP_FULL="$LATEST_SNAP"
    local snap_name="${SNAP_FULL#*@}"
    # An old snapshot is worth pointing out (the backup then contains an old
    # state), but the backup itself still succeeds - so this must not turn the
    # whole run into a FAILED one.
    if is_snap_stale "$SNAP_FULL"; then
        log_warn "  Snapshot older than 24h (${snap_name} on ${dataset}) - backing up that state"
    fi
    return 0
}


get_restic_json() {
    local output
    output=$(restic "$@" --json 2>/dev/null) || {
        log_error "restic $* failed"
        return 1
    }

    if ! echo "$output" | jq . >/dev/null 2>&1; then
        log_error "Invalid JSON from restic ($*)"
        return 1
    fi

    echo "$output"
}


find_last_full() {
    local disk_tag="$1"
    FULL_RESTIC_ID=""
    BASE_SNAP_NAME=""
    FULL_TIME=""
    local snapshots_json
    snapshots_json=$(get_restic_json snapshots --tag "${disk_tag},full") || return 1
    [[ -z "$snapshots_json" || "$snapshots_json" == "[]" ]] && return 1
    local _parsed
    _parsed=$(echo "$snapshots_json" | jq -r '
        sort_by(.time) | last as $s |
        [$s.short_id // "", $s.time // "",
         ([$s.tags // [] | .[] | select(startswith("base-snap=")) | sub("^base-snap="; "")][0] // "")] |
        @tsv
    ' 2>/dev/null) || return 1
    IFS=$'\t' read -r FULL_RESTIC_ID FULL_TIME BASE_SNAP_NAME <<< "$_parsed"
    [[ -n "$FULL_RESTIC_ID" && -n "$BASE_SNAP_NAME" ]]
}

is_full_too_old() {
    [[ -z "$FULL_TIME" ]] && return 0
    local full_epoch now_epoch age
    full_epoch=$(date -d "$FULL_TIME" +%s 2>/dev/null) || return 0
    now_epoch=$(date +%s)
    age=$(( (now_epoch - full_epoch) / 86400 ))
    [[ $age -ge $FULL_MAX_AGE_DAYS ]]
}

base_snap_exists() {
    local dataset="$1"
    local base_full_snap="${dataset}@${BASE_SNAP_NAME}"
    zfs list -t snapshot -H -o name "$base_full_snap" &>/dev/null
}

full_restic_snap_exists() {
    local snap_id="$1"
    [[ -z "$snap_id" ]] && return 1
    restic snapshots --json "$snap_id" 2>/dev/null | jq -e 'length > 0' >/dev/null 2>&1
}

# Write a zfs send stream to restic. $5 = base snapshot; empty = full backup.
# Full and differential differ in the -i flag and in three tags only, so both
# go through one function - including the identical cleanup of a stream that
# restic accepted but zfs send did not finish writing.
backup_zfs_stream() {
    local vmid="$1" disk_key="$2" dataset="$3" current_snap="$4" base_snap="${5:-}"
    local disk_tag="vm-${vmid}-${disk_key}"
    local filename="vm-${vmid}-${disk_key}.zfs"
    local kind="full"
    local -a send_args=(-w) tags=()

    if [[ -n "$base_snap" ]]; then
        kind="differential"
        send_args+=(-i "${dataset}@${base_snap}")
        tags=(--tag "differential" --tag "parent-snap=${base_snap}" \
              --tag "parent-id=${FULL_RESTIC_ID}")
    else
        tags=(--tag "full" --tag "base-snap=${current_snap}")
    fi

    local _err
    _err=$(mktemp) || _err=""
    set +e +o pipefail
    zfs send "${send_args[@]}" "${SNAP_FULL}" 2>/dev/null | restic backup \
        --stdin --stdin-filename "$filename" --compression "$RESTIC_COMPRESSION_SEND" \
        --tag "vm-${vmid}" --tag "$disk_tag" "${tags[@]}" \
        "${NAME_TAG_FLAGS[@]}" "${CONFIG_ID_TAG_FLAGS[@]}" &>"${_err:-/dev/null}"
    local _ps=("${PIPESTATUS[@]}")
    set -e -o pipefail
    if [[ -n "$_err" ]]; then
        log_restic_output "${kind} ${disk_key}" "$(cat "$_err" 2>/dev/null)"
        rm -f "$_err"
    fi
    local _send_rc=${_ps[0]:-1}
    local _restic_rc=${_ps[1]:-1}

    if [[ "$_restic_rc" -eq 0 && "$_send_rc" -eq 0 ]]; then
        log_info "  OK: ${disk_key} ${kind} (${dataset})"
        TOTAL_DISKS=$((TOTAL_DISKS+1))
        if [[ -n "$base_snap" ]]; then
            TOTAL_DIFFS=$((TOTAL_DIFFS+1))
        else
            TOTAL_FULLS=$((TOTAL_FULLS+1))
        fi
        return 0
    fi

    if [[ "$_restic_rc" -eq 0 ]]; then
        # restic stored a truncated stream - it would restore as a broken image
        log_warn "  Partial: ${disk_key} ${kind} (zfs send exit=${_send_rc}) - removing incomplete backup"
        local _bad
        _bad=$(restic snapshots --tag "${disk_tag},${kind}" --json 2>/dev/null \
            | jq -r 'sort_by(.time) | last | .short_id // empty' 2>/dev/null) || true
        if [[ -n "$_bad" ]]; then
            restic forget "$_bad" &>/dev/null \
                || log_warn "  Could not auto-remove ${_bad}, remove manually: restic forget ${_bad}"
        fi
    else
        log_error "  Failed: ${filename} (restic=${_restic_rc}, send=${_send_rc})"
    fi
    return 1
}


# Decides for a disk already prepared by prepare_snapshot() whether a full
# backup would be needed. Side effect: sets FULL_RESTIC_ID / BASE_SNAP_NAME /
# FULL_TIME (via find_last_full) and reports the reason in DISK_FULL_REASON.
# Return: 0 = full needed, 1 = differential possible.
# EFI and TPM state volumes are ALWAYS backed up full. They are tiny, they
# change abruptly and a differential on them gains nothing. They deliberately
# do NOT trigger the all-or-nothing rule - the rest of the run stays
# differential.
is_always_full_disk() {
    case "$1" in
        efidisk*|tpmstate*) return 0 ;;
        *)                  return 1 ;;
    esac
}

DISK_FULL_REASON=""
disk_needs_full() {
    local disk_tag="$1" dataset="$2"
    DISK_FULL_REASON=""
    FULL_RESTIC_ID=""
    BASE_SNAP_NAME=""
    FULL_TIME=""
    if ! find_last_full "$disk_tag"; then
        DISK_FULL_REASON="no previous full in repo"; return 0
    elif is_full_too_old; then
        DISK_FULL_REASON="last full older than ${FULL_MAX_AGE_DAYS} days"; return 0
    elif ! base_snap_exists "$dataset"; then
        DISK_FULL_REASON="base snapshot ${BASE_SNAP_NAME} missing"; return 0
    elif ! full_restic_snap_exists "$FULL_RESTIC_ID"; then
        DISK_FULL_REASON="referenced full restic snapshot ${FULL_RESTIC_ID} no longer exists in repo"; return 0
    fi
    return 1
}

backup_zfs_disk() {
    local vmid="$1" disk_key="$2" dataset="$3"
    local disk_tag="vm-${vmid}-${disk_key}"
    if ! prepare_snapshot "$dataset"; then
        return 1
    fi
    local current_snap="${SNAP_FULL#*@}"
    log_info "  Backing up ${disk_key}: ZFS snapshot ${SNAP_FULL}"

    # Decided per disk. EFI and TPM state volumes are always full: they are
    # tiny, they change abruptly and a differential on them gains nothing.
    local base_snap="" reason=""
    if is_always_full_disk "$disk_key"; then
        # find_last_full only to fill the tag variables; the result is moot
        find_last_full "$disk_tag" >/dev/null 2>&1 || true
        reason="EFI/TPM state disks are always backed up full"
    elif disk_needs_full "$disk_tag" "$dataset"; then
        reason="$DISK_FULL_REASON"
    else
        base_snap="$BASE_SNAP_NAME"
        if [[ "$base_snap" == "$current_snap" ]]; then
            log_info "  Skip: snapshot unchanged since last full (${current_snap})"
            return 0
        fi
    fi

    if $RESTIC_DRY_RUN; then
        if [[ -z "$base_snap" ]]; then
            log_info "  [DRY-RUN] FULL (${reason}): zfs send -w ${SNAP_FULL} | restic --stdin"
        else
            log_info "  [DRY-RUN] DIFFERENTIAL: zfs send -w -i @${base_snap} ${SNAP_FULL} | restic --stdin"
        fi
        return 0
    fi

    if [[ -z "$base_snap" ]]; then
        log_info "  Full backup (${reason})"
        backup_zfs_stream "$vmid" "$disk_key" "$dataset" "$current_snap"
        return $?
    fi

    log_info "  Differential: ${base_snap} -> ${current_snap}"
    backup_zfs_stream "$vmid" "$disk_key" "$dataset" "$current_snap" "$base_snap" && return 0
    log_warn "  Differential failed -> falling back to a full backup"
    backup_zfs_stream "$vmid" "$disk_key" "$dataset" "$current_snap"
}

backup_lxc_fs() {
    local ctid="$1" key="$2" dataset="$3"
    local mnt="/tmp/restic-snap-ct${ctid}-${key}"
    if ! prepare_snapshot "$dataset"; then
        return 1
    fi
    log_info "  Backing up CT ${ctid} ${key}: ZFS snapshot ${SNAP_FULL}"

    if $RESTIC_DRY_RUN; then
        log_info "  [DRY-RUN] mount ${SNAP_FULL} ro -> restic backup ${mnt}"
        return 0
    fi
    for stale in /tmp/restic-snap-*; do
        [[ -d "$stale" ]] || continue
        if mountpoint -q "$stale" 2>/dev/null; then
            umount "$stale" 2>/dev/null || true
        fi
        rmdir "$stale" 2>/dev/null || true
    done
    mkdir -p "$mnt"
    if ! mount -t zfs -o ro "${SNAP_FULL}" "$mnt" 2>/dev/null; then
        log_error "  Failed to mount ${SNAP_FULL}"
        rmdir "$mnt" 2>/dev/null || true
        return 1
    fi
    local rc=1 _out=""
    _out=$( cd "$mnt" && restic backup . "${LXC_EXCLUDE_FLAGS[@]}" \
        --compression "$RESTIC_COMPRESSION_FILES" \
        --tag "ct-${ctid}" --tag "lxc" "${NAME_TAG_FLAGS[@]}" "${CONFIG_ID_TAG_FLAGS[@]}" 2>&1 ) && rc=0 || rc=$?
    log_restic_output "CT ${ctid} ${key}" "$_out"
    if [[ $rc -eq 0 ]]; then
        log_info "  OK: ${key} (${dataset})"
        TOTAL_DISKS=$((TOTAL_DISKS+1))
    else
        log_error "  Failed: ct-${ctid}-${key} (restic exit=${rc})"
    fi
    umount "$mnt" 2>/dev/null || true
    rmdir "$mnt" 2>/dev/null || true
}

# LXC rootfs on a directory storage without ZFS: back up file based straight
# from the running directory (no snapshot -> no point-in-time consistency).
backup_lxc_dir() {
    local ctid="$1" key="$2" dir="$3"
    if [[ ! -d "$dir" ]]; then
        log_error "  ${key}: directory ${dir} not found"
        return 1
    fi
    if $RESTIC_DRY_RUN; then
        log_info "  [DRY-RUN] restic backup ${dir} (LXC file-based, no snapshot)"
        return 0
    fi
    log_warn "  CT ${ctid} ${key}: no snapshot possible (${dir}) - live backup"
    local rc=1 _out=""
    _out=$( cd "$dir" && restic backup . "${LXC_EXCLUDE_FLAGS[@]}" \
        --compression "$RESTIC_COMPRESSION_FILES" \
        --tag "ct-${ctid}" --tag "lxc" "${NAME_TAG_FLAGS[@]}" "${CONFIG_ID_TAG_FLAGS[@]}" 2>&1 ) && rc=0 || rc=$?
    log_restic_output "CT ${ctid} ${key}" "$_out"
    if [[ $rc -eq 0 ]]; then
        log_info "  OK: ${key} (${dir})"
        TOTAL_DISKS=$((TOTAL_DISKS+1))
    else
        log_error "  Failed: ct-${ctid}-${key} (restic exit=${rc})"
    fi
}

backup_config() {
    local id="$1" subdir="$2" prefix="$3"
    local node
    node=$(hostname -s)
    local api_path
    if [[ "$subdir" == "qemu-server" ]]; then
        api_path="/nodes/${node}/qemu/${id}/config"
    else
        api_path="/nodes/${node}/lxc/${id}/config"
    fi
    if $RESTIC_DRY_RUN; then
        log_info "  [DRY-RUN] pvesh get ${api_path} | restic --stdin"
        return 0
    fi
    local _err
    _err=$(mktemp) || _err=""
    set +e +o pipefail
    pvesh get "$api_path" --output-format json 2>/dev/null | restic backup \
        --stdin --stdin-filename "${prefix}-${id}.conf" \
        --compression "$RESTIC_COMPRESSION_FILES" \
        --tag "${prefix}-${id}" --tag "config" "${NAME_TAG_FLAGS[@]}" &>"${_err:-/dev/null}"
    local _ps=("${PIPESTATUS[@]}")
    set -e -o pipefail
    if [[ -n "$_err" ]]; then
        log_restic_output "config ${prefix}-${id}" "$(cat "$_err" 2>/dev/null)"
        rm -f "$_err"
    fi
    local _pvesh_rc=${_ps[0]:-1}
    local _restic_rc=${_ps[1]:-1}
    if [[ "$_restic_rc" -eq 0 && "$_pvesh_rc" -eq 0 ]]; then
        log_info "  OK: config"
    elif [[ "$_restic_rc" -eq 0 ]]; then
        log_warn "  Partial: config (pvesh exit=${_pvesh_rc})"
    else
        log_error "  Failed: config for ${prefix}-${id}"
        CURRENT_CONFIG_ID=""
        return 0
    fi
    # Read the newest config snapshot ID - it is the group identifier for
    # all disk backups of this backup run
    CURRENT_CONFIG_ID=$(restic snapshots --tag "${prefix}-${id},config" --json 2>/dev/null \
        | jq -r 'sort_by(.time) | last | .short_id // empty' 2>/dev/null) || CURRENT_CONFIG_ID=""
    [[ -n "$CURRENT_CONFIG_ID" ]] && log_info "  Config-Snapshot-ID: ${CURRENT_CONFIG_ID}"
}

backup_local_dir() {
    local dir="$1"
    local dirname="${dir%/}"
    dirname="${dirname##*/}"
    if [[ ! -d "$dir" ]]; then
        log_error "  Directory not found: ${dir}"
        return 1
    fi
    if $RESTIC_DRY_RUN; then
        log_info "  [DRY-RUN] restic backup ${dir} ${LOCAL_EXCLUDE_FLAGS[*]:-}"
        return 0
    fi
    log_info "  Backing up: ${dir}"
    local rc=0
    restic_logged "local ${dirname}" backup "$dir" "${LOCAL_EXCLUDE_FLAGS[@]}" \
        --compression "$RESTIC_COMPRESSION_FILES" \
        --tag "local" --tag "dir-${dirname}" || rc=$?
    if [[ $rc -eq 0 ]]; then
        log_info "  OK: ${dir}"
        TOTAL_DISKS=$((TOTAL_DISKS+1))
    else
        log_error "  Failed: ${dir} (restic exit=${rc})"
    fi
}

# Everything restic prints has to end up in LOG_OUTPUT, otherwise it only ever
# reaches the cron output and is missing from the notification mail - which is
# exactly where it is wanted when something went wrong.
#
# The severity stays with the caller, which knows what a non-zero exit code
# means in its context, so neither helper here touches a counter of its own.
#
# IMPORTANT: both must run in the CURRENT shell. The right-hand side of a pipe
# is a subshell, and an append to LOG_OUTPUT there would be discarded - which is
# why the piped backup calls capture into a file and drain it afterwards instead
# of piping into restic_logged.
log_restic_output() {
    local _label="$1" _line
    [[ -z "${2:-}" ]] && return 0
    while IFS= read -r _line; do
        [[ -n "$_line" ]] && rb_log "RESTIC" "${_label}: ${_line}"
    done <<< "$2"
    return 0
}

# For restic calls that are NOT part of a pipe. Returns restic's exit code.
restic_logged() {
    local _label="$1"; shift
    local _out _rc=0
    _out=$(restic "$@" 2>&1) || _rc=$?
    log_restic_output "$_label" "$_out"
    return "$_rc"
}

restic() {
    local restic_opts=("--repo" "$RESTIC_REPOSITORY" "--password-file" "$RESTIC_PASSWORDFILE")
    if $RESTIC_VERBOSE; then
        restic_opts+=("--verbose")
    else
        restic_opts+=("--quiet")
    fi
    command restic "${restic_opts[@]}" "$@"
}

# restic keeps a lock inside the repository while it works. A run that was
# killed (power loss, OOM, Ctrl+C, ERR trap) leaves it behind and every later
# run fails with "repository is already locked".
# The flock at the top of this script already guarantees that only one instance
# runs at a time, so a lock found here is stale by definition. Plain
# 'restic unlock' only removes stale locks; if an exclusive lock survives that,
# it is removed explicitly.
clear_stale_repo_lock() {
    $RESTIC_DRY_RUN && return 0
    # No repo yet (first run) or no locks at all: nothing to do.
    restic list locks 2>/dev/null | grep -q . || return 0

    log_warn "Lock found in the repository (previous run aborted?) - removing it"
    restic unlock &>/dev/null || true
    if ! restic list locks 2>/dev/null | grep -q .; then
        log_info "Stale lock removed"
        return 0
    fi
    if restic unlock --remove-all &>/dev/null; then
        log_info "Exclusive lock removed"
    else
        log_error "restic unlock failed - remove it manually: restic unlock --remove-all"
    fi
}





# Warning after the run if less than DISK_SPACE_WARN_PERCENT is free.
check_disk_space() {
    local free_bytes total_bytes threshold_bytes
    free_bytes=$(get_free_bytes) || return 0
    total_bytes=$(get_total_bytes) || return 0
    threshold_bytes=$(( total_bytes * DISK_SPACE_WARN_PERCENT / 100 ))

    log_info "Disk space: $(human_bytes "$free_bytes") free of $(human_bytes "$total_bytes") (warning threshold ${DISK_SPACE_WARN_PERCENT}% = $(human_bytes "$threshold_bytes"))"

    if [[ "$free_bytes" -lt "$threshold_bytes" ]]; then
        log_warn "WARNING: free space on the target drive is low, only $(human_bytes "$free_bytes") left (below ${DISK_SPACE_WARN_PERCENT}%)"
    fi
}

# ================== Post-Backup Integrity Check ==================

bytes_to_restic_size() {
    local bytes="$1"
    if [[ "$bytes" -ge 1073741824 ]]; then
        echo "$(( (bytes + 1073741823) / 1073741824 ))G"
    elif [[ "$bytes" -ge 1048576 ]]; then
        echo "$(( (bytes + 1048575) / 1048576 ))M"
    elif [[ "$bytes" -ge 1024 ]]; then
        echo "$(( (bytes + 1023) / 1024 ))K"
    else
        echo "${bytes}"
    fi
}

human_bytes() {
    local bytes="$1"
    if [[ "$bytes" -ge 1073741824 ]]; then
        awk -v b="$bytes" 'BEGIN{printf "%.1f GiB", b/1073741824}'
    elif [[ "$bytes" -ge 1048576 ]]; then
        awk -v b="$bytes" 'BEGIN{printf "%.1f MiB", b/1048576}'
    elif [[ "$bytes" -ge 1024 ]]; then
        awk -v b="$bytes" 'BEGIN{printf "%.1f KiB", b/1024}'
    else
        echo "${bytes} B"
    fi
}

# The rotation counter lives ON THE MEDIUM, next to the repository it belongs
# to. With several drives in rotation each one carries its own repository, and a
# single counter on the host would advance for whichever drive happens to be
# plugged in - leaving every drive with permanent gaps instead of giving each
# one full coverage.
CHECK_ROTATE_STATE=""
CHECK_ROTATE_N=1

# Determines the group for this run and the path of the counter file.
# Sets CHECK_ROTATE_N / CHECK_ROTATE_STATE instead of echoing the number: a
# command substitution would run this in a subshell and the path would be lost
# before _check_rotate_advance could write to it.
_check_rotate_current() {
    CHECK_ROTATE_STATE="${RESTIC_BASEFOLDER%/}/.restic-check-rotate"
    local n=1
    if [[ -r "$CHECK_ROTATE_STATE" ]]; then
        n=$(head -1 "$CHECK_ROTATE_STATE" 2>/dev/null | tr -cd '0-9') || n=""
    fi
    # Also catches a RESTIC_CHECK_ROTATE_RUNS that was lowered since last time.
    [[ "$n" =~ ^[0-9]+$ ]] || n=1
    [[ "$n" -ge 1 && "$n" -le "$RESTIC_CHECK_ROTATE_RUNS" ]] || n=1
    CHECK_ROTATE_N="$n"
    return 0
}

_check_rotate_advance() {
    local n=$(( $1 + 1 ))
    [[ "$n" -gt "$RESTIC_CHECK_ROTATE_RUNS" ]] && n=1
    echo "$n" > "$CHECK_ROTATE_STATE" 2>/dev/null \
        || log_warn "  Cannot write ${CHECK_ROTATE_STATE} - group ${1} will be read again next run"
    return 0
}

run_post_backup_check() {
    [[ "$RESTIC_CHECK_AFTER_BACKUP" != "true" ]] && return 0
    $RESTIC_DRY_RUN && { log_info "Skipping check (dry-run)"; return 0; }

    log_info "=== Post-backup integrity check ==="

    local _check_rc=0

    if [[ "$RESTIC_CHECK_ROTATE" == "true" ]]; then
        # Deliberately runs even when this run wrote nothing: the point is to
        # walk the whole repository over time, not to check the fresh data.
        # Deduplication means an old blob is referenced again without ever being
        # read, so a backup on its own never revalidates what is already there.
        local n subset
        _check_rotate_current
        n="$CHECK_ROTATE_N"
        subset="${n}/${RESTIC_CHECK_ROTATE_RUNS}"
        log_info "Reading back group ${subset} of the pack files"
        log_info "  (every pack is verified once per ${RESTIC_CHECK_ROTATE_RUNS} runs)"

        restic_logged "check" check --read-data-subset="$subset" || _check_rc=$?

        # Exit 1 means the check ran and found problems - the group WAS read, so
        # the rotation moves on. Anything else means it never got that far.
        if [[ "$_check_rc" -le 1 ]]; then
            _check_rotate_advance "$n"
        else
            log_warn "  Check did not complete - group ${subset} is read again next run"
        fi
    else
        local data_added="$TOTAL_DATA_ADDED"
        if [[ "$data_added" -le 0 ]]; then
            log_info "No new data written this run, skipping check"
            return 0
        fi

        local max_check_bytes=$(( RESTIC_CHECK_MAX_GB * 1024 * 1024 * 1024 ))
        local check_bytes="$data_added"
        [[ "$check_bytes" -gt "$max_check_bytes" ]] && check_bytes="$max_check_bytes"

        local check_size_str
        check_size_str=$(bytes_to_restic_size "$check_bytes")

        log_info "Data written this run: $(human_bytes "$data_added")"
        log_info "Checking a random ${check_size_str} (max: ${RESTIC_CHECK_MAX_GB} GiB)"

        restic_logged "check" check --read-data-subset="$check_size_str" || _check_rc=$?
    fi

    if [[ "$_check_rc" -eq 0 ]]; then
        log_info "Integrity check passed"
    else
        log_error "Integrity check FAILED (exit=${_check_rc}) - repository may be corrupted"
    fi
    return 0
}

# ================== Space management for full backups ==================

# Determine the free space on the target drive, in bytes.
get_free_bytes() {
    local avail block_size
    avail=$(stat -f -c '%a' "$RESTIC_BASEFOLDER" 2>/dev/null) || return 1
    block_size=$(stat -f -c '%S' "$RESTIC_BASEFOLDER" 2>/dev/null) || block_size=512
    echo $(( avail * block_size ))
}

# Total size of the target drive, in bytes.
get_total_bytes() {
    local total block_size
    total=$(stat -f -c '%b' "$RESTIC_BASEFOLDER" 2>/dev/null) || return 1
    block_size=$(stat -f -c '%S' "$RESTIC_BASEFOLDER" 2>/dev/null) || block_size=512
    echo $(( total * block_size ))
}

# ================== Space calculation helpers ==================

# Determine the ZFS space actually in use (used, bytes) of a dataset/zvol.
zfs_used_bytes() {
    local dataset="$1" v
    v=$(zfs get -Hp -o value used "$dataset" 2>/dev/null) || return 1
    [[ "$v" =~ ^[0-9]+$ ]] || return 1
    echo "$v"
}

# Estimated stream size of a 'zfs send' in bytes, WITHOUT reading data.
# $1 dataset, $2 target snapshot name, $3 base snapshot name (empty = full).
# The flags must match those of the real send (-w), otherwise the estimate
# drifts. ZFS deliberately returns an estimate, not an exact value.
zfs_send_size() {
    local dataset="$1" target_snap="$2" base_snap="${3:-}"
    local args=(-w -n -P) out=""
    [[ -n "$base_snap" ]] && args+=(-i "${dataset}@${base_snap}")
    set +e +o pipefail
    out=$(zfs send "${args[@]}" "${dataset}@${target_snap}" 2>/dev/null \
          | awk '$1=="size"{print $2; exit}')
    set -e -o pipefail
    [[ "$out" =~ ^[0-9]+$ ]] || return 1
    echo "$out"
}

# Space occupied by a single restic snapshot (bytes).
# With compression enabled raw-data tends to overestimate - which is the safe
# direction for space planning.
snapshot_repo_bytes() {
    local snap_id="$1" v=""
    set +e +o pipefail
    v=$(restic stats --mode raw-data --json "$snap_id" 2>/dev/null \
        | grep -x '^{.*}' | jq -r '.total_size // 0' 2>/dev/null)
    set -e -o pipefail
    [[ "$v" =~ ^[0-9]+$ ]] || return 1
    echo "$v"
}

# Size of a directory tree in bytes (only needed when there is no backup of
# this tag yet - which is why it is called that late on purpose).
dir_size_bytes() {
    local v
    v=$(du -sb "$1" 2>/dev/null | awk '{print $1; exit}') || return 1
    [[ "$v" =~ ^[0-9]+$ ]] || return 1
    echo "$v"
}

# Estimate the space required by a file based or RAW backup.
#   $1        = tag filter (comma separated, restic semantics = AND)
#   $2..$n    = command returning the FULL source size (lazy - only executed
#               when there is no backup for this tag yet)
# Present     -> ESTIMATED_BACKUP_INCREASE_PERCENTAGE % of the space in use
# Not present -> full source size, without assuming compression
estimate_backup_need() {
    local tag_csv="$1"; shift
    local last_id repo_bytes full
    last_id=$(restic snapshots --tag "$tag_csv" --json 2>/dev/null \
        | jq -r 'sort_by(.time) | last | .short_id // empty' 2>/dev/null) || last_id=""
    if [[ -n "$last_id" ]]; then
        repo_bytes=$(snapshot_repo_bytes "$last_id") || repo_bytes=0
        echo $(( repo_bytes * ESTIMATED_BACKUP_INCREASE_PERCENTAGE / 100 ))
        return 0
    fi
    full=$("$@" 2>/dev/null) || full=0
    [[ "$full" =~ ^[0-9]+$ ]] || full=0
    echo "$full"
}

# =================== Snapshot index ===================
# One pass over 'restic snapshots', chronologically sorted (oldest first).
# SNAP_IDS keeps that order, so "the newest snapshot of a group" is simply the
# last one seen while iterating, and the backup days come out in order.
declare -a SNAP_IDS=()
declare -A SNAP_TIME=()
declare -A SNAP_DAY=()
declare -A SNAP_TAGS=()

load_snapshot_index() {
    SNAP_IDS=(); SNAP_TIME=(); SNAP_DAY=(); SNAP_TAGS=()
    local rows _id _time _tags
    rows=$(restic snapshots --json 2>/dev/null | jq -r '
        sort_by(.time) | .[]
        | [ (.short_id // ""), ((.time // "")[0:16] | sub("T"; " ")),
            ((.tags // []) | join(",")) ]
        | @tsv
    ' 2>/dev/null) || return 1
    [[ -z "$rows" ]] && return 0
    while IFS=$'\t' read -r _id _time _tags; do
        [[ -z "$_id" ]] && continue
        SNAP_IDS+=("$_id")
        SNAP_TIME["$_id"]="$_time"
        SNAP_DAY["$_id"]="${_time:0:10}"
        SNAP_TAGS["$_id"]=",${_tags},"
    done <<< "$rows"
    return 0
}

snap_has_tag() { [[ "${SNAP_TAGS[$1]:-}" == *",$2,"* ]]; }

# Value of the tag beginning with $2 (prefix including '='), "" if absent.
snap_tag_value() {
    local _all="${SNAP_TAGS[$1]:-}" _v
    _v="${_all#*,$2}"
    [[ "$_v" == "$_all" ]] && { echo ""; return 1; }
    echo "${_v%%,*}"
}

# Volume tag (vm-101-scsi0, ct-100-rootfs) of a snapshot, "" if it has none.
snap_disk_tag() {
    [[ "${SNAP_TAGS[$1]:-}" =~ ,((vm|ct)-[0-9]+-[^,]+), ]] && { echo "${BASH_REMATCH[1]}"; return 0; }
    echo ""; return 1
}

# efidisk*/tpmstate* are backed up full on EVERY run - they never carry a
# machine's history on their own.
snap_is_always_full_volume() {
    local _d
    _d=$(snap_disk_tag "$1") || return 1
    is_always_full_disk "${_d##*-}"
}

# Group of a self-contained backup (LXC, LOCAL_DIRS, RAW) - one line of
# history per group. Empty for fulls, differentials and config snapshots.
snap_group_key() {
    local _all="${SNAP_TAGS[$1]:-}" _k=""
    if snap_has_tag "$1" lxc || snap_has_tag "$1" raw; then
        _k=$(snap_disk_tag "$1") || _k=""
        [[ -z "$_k" && "$_all" =~ ,((vm|ct)-[0-9]+), ]] && _k="${BASH_REMATCH[1]}"
    elif snap_has_tag "$1" local; then
        [[ "$_all" =~ ,(dir-[^,]+), ]] && _k="${BASH_REMATCH[1]}"
    fi
    echo "$_k"
    [[ -n "$_k" ]]
}

# Backup type and owner of a snapshot, for log lines.
snap_kind() {
    snap_has_tag "$1" raw          && { echo "RAW";    return 0; }
    snap_has_tag "$1" full         && { echo "FULL";   return 0; }
    snap_has_tag "$1" differential && { echo "DIFF";   return 0; }
    snap_has_tag "$1" config       && { echo "CONFIG"; return 0; }
    snap_has_tag "$1" lxc          && { echo "LXC-FS"; return 0; }
    snap_has_tag "$1" local        && { echo "LOCAL";  return 0; }
    echo "?"
}
snap_owner() {
    [[ "${SNAP_TAGS[$1]:-}" =~ ,((vm|ct)-[0-9]+), ]] && { echo "${BASH_REMATCH[1]}"; return 0; }
    echo "-"
}

# Volume resp. directory a snapshot belongs to, as shown in the DISK column.
snap_disk_column() {
    local _d
    _d=$(snap_disk_tag "$1") || _d=""
    [[ -n "$_d" ]] && { echo "${_d#*-*-}"; return 0; }
    [[ "${SNAP_TAGS[$1]:-}" =~ ,dir-([^,]+), ]] && { echo "${BASH_REMATCH[1]}"; return 0; }
    echo "-"
}

# Restore size of a snapshot in bytes.
snap_restore_size() {
    local v
    v=$(restic stats --mode restore-size --json "$1" 2>/dev/null \
        | jq -r '.total_size // 0' 2>/dev/null) || v=0
    [[ "$v" =~ ^[0-9]+$ ]] || v=0
    echo "$v"
}

# Newest backup per LXC/LOCAL/RAW group. Those are never dropped while
# reducing - without the guard a run could throw away the only container
# backup while VM history survives. Only the final wipe touches them.
declare -A NEWEST_OF_GROUP=()
build_newest_of_group() {
    NEWEST_OF_GROUP=()
    [[ ${#SNAP_IDS[@]} -eq 0 ]] && return 0
    local _id _k
    for _id in "${SNAP_IDS[@]}"; do
        _k=$(snap_group_key "$_id") || continue
        NEWEST_OF_GROUP["$_k"]="$_id"
    done
    return 0
}


# Forget the given ids (protected configs excluded), clean up what that leaves
# behind and prune. $1 is the log label. Return 1 when nothing was removed.
# Forget the given ids and prune. $1 is the log label.
# Return 1 when nothing was removed; PRUNE_LAST_ERROR tells a failed removal
# apart from "there was nothing left to remove".
PRUNE_LAST_ERROR=false
forget_and_prune() {
    local label="$1"; shift
    [[ $# -eq 0 ]] && return 1
    log_warn "  ${label}: removing $# snapshot(s)"
    if ! restic_logged "forget" forget "$@"; then
        log_error "  forget failed"
        PRUNE_LAST_ERROR=true
        return 1
    fi
    restic_logged "prune" prune || log_warn "  restic prune failed"
    return 0
}

# ---------------------------------------------------------------------------
# Pre-backup pruning. REQUIRED_BYTES comes from the planning phase; if it does
# not fit, whole backup DAYS are released, oldest first:
#
#   0. a disk listed in ZFS_SEND_DISKS is never released here at all. It is on
#      zfs send because re-reading it whole is expensive, so dropping its chain
#      to win space would force exactly that expensive full on the next run.
#      If only such backups are left, the run says so and gives up rather than
#      releasing them - see ensure_space_available().
#   1. everything else written on that day goes - except the newest backup of a
#      machine or directory, which is never the thing to throw away
#   2. a full that goes takes its differentials with it, whichever day they
#      were written on: without their full they restore nothing
#   3. the always-full volumes (EFI/TPM state) of a run go once no real backup
#      of that run survives - on their own they restore nothing, but as long as
#      ANY disk of the run is still there they are needed to restore it. They
#      therefore follow their run rather than the day they were written on
#   4. config snapshots go once no surviving backup refers to them any more
#   5. still not enough? next day
#
# Disks listed in ZFS_SEND_DISKS are held back from all of this in a first pass.
# They were chosen deliberately because re-reading them as a RAW image is
# expensive, so destroying their chain is the costliest thing this could do.
# Only when releasing days frees nothing anywhere else does a second pass
# include them too - which is still better than the repository wipe below.
#
# Self-contained backups (RAW, LXC, LOCAL_DIRS) of those later days are never
# touched by 2. and 3. - they have no chain that could break.
# The reasoning behind all of it is documented in config.cfg.
# ---------------------------------------------------------------------------

# The distinct backup days, oldest first.
snapshot_days() {
    local _id _last=""
    for _id in ${SNAP_IDS[@]+"${SNAP_IDS[@]}"}; do
        [[ "${SNAP_DAY[$_id]}" == "$_last" ]] && continue
        _last="${SNAP_DAY[$_id]}"
        printf '%s\n' "$_last"
    done
    return 0
}

# Was this snapshot's disk deliberately put on zfs send via ZFS_SEND_DISKS?
snap_is_chosen_zfs_disk() {
    [[ -n "$ZFS_SEND_DISKS" ]] || return 1
    local _t
    _t=$(snap_disk_tag "$1") || return 1
    [[ "$_t" == vm-* ]] || return 1
    _t="${_t#vm-}"            # 100-scsi1
    _t="${_t/-/:}"            # 100:scsi1 - disk keys never contain a dash
    [[ ",${ZFS_SEND_DISKS}," == *",${_t},"* ]]
}

# True when no full backup of this volume survives the current doomed set.
# Reads the caller's 'doomed' array - bash scopes locals dynamically, and
# passing a whole associative array as an argument is not worth the noise.
_last_full_gone() {
    local _v="$1" _i _t
    for _i in "${SNAP_IDS[@]}"; do
        [[ -n "${doomed[$_i]:-}" ]] && continue
        snap_has_tag "$_i" full || continue
        _t=$(snap_disk_tag "$_i") || continue
        [[ "$_t" == "$_v" ]] && return 1
    done
    return 0
}

# Release the oldest day that still has anything to release.
# Return 1 when no day is left that would free something.
PRUNE_PINNED_SKIPPED=0
prune_oldest_day() {
    local _day _id _p _cfg
    local -a days=()
    PRUNE_PINNED_SKIPPED=0
    mapfile -t days < <(snapshot_days)

    for _day in ${days[@]+"${days[@]}"}; do
        local -A doomed=() live_runs=() still_used=()

        # 1) everything written that day. Always-full volumes (EFI/TPM state)
        #    are left to step 3: they follow their run, not the calendar, and
        #    they are far too small for their removal to free anything.
        for _id in "${SNAP_IDS[@]}"; do
            [[ "${SNAP_DAY[$_id]}" == "$_day" ]] || continue
            snap_has_tag "$_id" config && continue
            snap_is_always_full_volume "$_id" && continue
            _is_newest_of_group "$_id" && continue
            # A disk from ZFS_SEND_DISKS is never released here. It is on zfs
            # send because re-reading it whole is expensive, and dropping its
            # chain would force exactly that expensive full on the next run.
            if snap_is_chosen_zfs_disk "$_id"; then
                PRUNE_PINNED_SKIPPED=$((PRUNE_PINNED_SKIPPED+1))
                continue
            fi
            doomed["$_id"]=1
        done

        # 2) differentials of a doomed full, on any day
        for _id in "${SNAP_IDS[@]}"; do
            snap_has_tag "$_id" differential || continue
            [[ -n "${doomed[$_id]:-}" ]] && continue
            _p=$(snap_tag_value "$_id" "parent-id=") || _p=""
            [[ -n "$_p" && -n "${doomed[$_p]:-}" ]] && doomed["$_id"]=1
        done

        [[ ${#doomed[@]} -eq 0 ]] && continue

        # 3) EFI/TPM state volumes of a run that has nothing left to restore.
        #    config-id identifies one run of one machine, so this stays within
        #    the machine and never reaches an unrelated backup. As long as any
        #    real disk of the run survives - a protected one, or a RAW image of
        #    a second disk - the EFI volume is still needed and stays.
        for _id in "${SNAP_IDS[@]}"; do
            [[ -n "${doomed[$_id]:-}" ]] && continue
            snap_has_tag "$_id" config && continue
            snap_is_always_full_volume "$_id" && continue
            _cfg=$(snap_tag_value "$_id" "config-id=") || _cfg=""
            [[ -n "$_cfg" ]] && live_runs["$_cfg"]=1
        done
        for _id in "${SNAP_IDS[@]}"; do
            [[ -n "${doomed[$_id]:-}" ]] && continue
            snap_is_always_full_volume "$_id" || continue
            _is_newest_of_group "$_id" && continue
            _cfg=$(snap_tag_value "$_id" "config-id=") || _cfg=""
            [[ -n "$_cfg" && -z "${live_runs[$_cfg]:-}" ]] && doomed["$_id"]=1
        done

        # 4) config snapshots nothing points at any more
        for _id in "${SNAP_IDS[@]}"; do
            [[ -n "${doomed[$_id]:-}" ]] && continue
            _cfg=$(snap_tag_value "$_id" "config-id=") || _cfg=""
            [[ -n "$_cfg" ]] && still_used["$_cfg"]=1
        done
        for _id in "${SNAP_IDS[@]}"; do
            snap_has_tag "$_id" config || continue
            [[ -n "${doomed[$_id]:-}" ]] && continue
            [[ -z "${still_used[$_id]:-}" ]] && doomed["$_id"]=1
        done

        # Losing the last full of a volume means the drive cannot hold even one
        # generation of it - a hardware problem, not a retention decision.
        local _dt
        for _id in "${!doomed[@]}"; do
            snap_has_tag "$_id" full || continue
            snap_is_always_full_volume "$_id" && continue
            _dt=$(snap_disk_tag "$_id") || continue
            if _last_full_gone "$_dt"; then
                log_warn "  Last full backup of ${_dt} is being released"
                _low_memory_warning
                break
            fi
        done

        forget_and_prune "Releasing backup day ${_day}" "${!doomed[@]}" || return 1
        return 0
    done
    return 1
}

_is_newest_of_group() {
    local _k
    _k=$(snap_group_key "$1") || return 1
    [[ "${NEWEST_OF_GROUP[$_k]:-}" == "$1" ]]
}

LOW_MEMORY_WARNED=false
_low_memory_warning() {
    $LOW_MEMORY_WARNED && return 0
    LOW_MEMORY_WARNED=true
    [[ "${LOW_MEMORY_WARNING^^}" == "TRUE" ]] && \
        log_warn "WARNING: target drive too small! A larger drive is required!"
    return 0
}

# Last resort: releasing days cannot free anything any more.
_wipe_repository() {
    load_snapshot_index || return 1
    [[ ${#SNAP_IDS[@]} -eq 0 ]] && return 1

    # Even the last resort keeps the disks from ZFS_SEND_DISKS - and the config
    # snapshots they need, or what survives could no longer be fully restored.
    local _id _c _kept
    local -A keep_cfg=()
    local -a _doomed=()
    for _id in "${SNAP_IDS[@]}"; do
        snap_is_chosen_zfs_disk "$_id" || continue
        _c=$(snap_tag_value "$_id" "config-id=") || _c=""
        [[ -n "$_c" ]] && keep_cfg["$_c"]=1
    done
    for _id in "${SNAP_IDS[@]}"; do
        snap_is_chosen_zfs_disk "$_id" && continue
        [[ -n "${keep_cfg[$_id]:-}" ]] && continue
        _doomed+=("$_id")
    done
    [[ ${#_doomed[@]} -eq 0 ]] && return 1

    _kept=$(( ${#SNAP_IDS[@]} - ${#_doomed[@]} ))
    if [[ "$_kept" -gt 0 ]]; then
        log_warn "  Wiping ${#_doomed[@]} snapshot(s) - ${_kept} from ZFS_SEND_DISKS are kept"
    else
        log_warn "  Wiping the remaining ${#_doomed[@]} snapshot(s) - nothing else can be released"
    fi
    restic_logged "forget (wipe)" forget "${_doomed[@]}" || return 1
    restic_logged "prune (wipe)" prune || log_warn "  restic prune (final) failed"
    return 0
}

# Measure the space once and compare it against the requirement.
# Return: 0 = enough free, 1 = too little. Sets SPACE_FREE_BYTES.
SPACE_FREE_BYTES=0
space_is_sufficient() {
    SPACE_FREE_BYTES=$(get_free_bytes) || return 0
    [[ "$SPACE_FREE_BYTES" -ge "$REQUIRED_BYTES" ]]
}

# Central space check / release BEFORE the backup.
ensure_space_available() {
    $FULL_SPACE_CHECK_DONE && return 0
    FULL_SPACE_CHECK_DONE=true
    $RESTIC_DRY_RUN && return 0

    local total_bytes guard=0
    log_info "=== Space check before backup ==="

    total_bytes=$(get_total_bytes) || total_bytes=0
    if ! SPACE_FREE_BYTES=$(get_free_bytes); then
        log_warn "  Cannot determine free space - check skipped"
        return 0
    fi
    log_info "  Required: $(human_bytes "$REQUIRED_BYTES")  |  Free: $(human_bytes "$SPACE_FREE_BYTES") / $(human_bytes "$total_bytes")"

    if [[ "$SPACE_FREE_BYTES" -ge "$REQUIRED_BYTES" ]]; then
        log_info "  Enough free space -> no pre-backup pruning needed"
        return 0
    fi

    log_warn "  Not enough free space -> pre-backup pruning"

    while ! space_is_sufficient; do
        guard=$((guard+1))
        [[ $guard -gt 1000 ]] && { log_error "  Pruning aborted (guard limit)"; break; }

        load_snapshot_index || { log_warn "  Cannot read the snapshot list - pruning aborted"; break; }
        build_newest_of_group

        log_info "  Free: $(human_bytes "$SPACE_FREE_BYTES") - still $(human_bytes "$REQUIRED_BYTES") required"

        prune_oldest_day && continue
        $PRUNE_LAST_ERROR && { log_error "  Pruning failed - stopping here"; break; }

        # Nothing left that releasing a day could free.
        _low_memory_warning
        if [[ "$PRUNE_PINNED_SKIPPED" -gt 0 ]]; then
            log_error "  ${PRUNE_PINNED_SKIPPED} backup(s) belong to disks from ZFS_SEND_DISKS and are"
            log_error "  deliberately not released - that is what putting them there means."
            log_error "  Free space by hand, use a larger drive, or take the disk out of"
            log_error "  ZFS_SEND_DISKS so it is stored as a RAW image again."
        fi
        _wipe_repository || true
        break
    done

    SPACE_FREE_BYTES=$(get_free_bytes) || SPACE_FREE_BYTES=0
    if [[ "$SPACE_FREE_BYTES" -ge "$REQUIRED_BYTES" ]]; then
        log_info "  Enough space after pruning: $(human_bytes "$SPACE_FREE_BYTES") free"
    else
        log_warn "  Still not enough space: $(human_bytes "$SPACE_FREE_BYTES") free, $(human_bytes "$REQUIRED_BYTES") required"
    fi
    return 0
}

# ================== Retention: forget + prune ==================
# The keep-daily/weekly/monthly policy does not know about our snapshot
# relations, so its removal list is corrected afterwards:
#   - a full that a KEPT differential still points at stays,
#   - and so does the config snapshot of everything that stays. A full without
#     its VM configuration would no longer be fully restorable, and the config
#     costs a few hundred bytes.
run_forget_prune() {
    log_info "=== Retention: forget + prune ==="
    log_info "Policy: keep-daily=${KEEP_DAILY} keep-weekly=${KEEP_WEEKLY} keep-monthly=${KEEP_MONTHLY}"

    local remove_ids
    remove_ids=$(restic forget \
        --keep-daily  "$KEEP_DAILY" \
        --keep-weekly "$KEEP_WEEKLY" \
        --keep-monthly "$KEEP_MONTHLY" \
        --group-by host,paths \
        --dry-run --json 2>/dev/null \
        | jq -r '.[].remove[]?.short_id // empty' 2>/dev/null) || {
        log_error "restic forget --dry-run failed"
        return 1
    }
    [[ -z "$remove_ids" ]] && { log_info "Nothing to forget"; return 0; }

    load_snapshot_index || { log_error "Cannot read the snapshot list"; return 1; }

    local -A doomed=() reason=()
    local _id _p _ref
    while IFS= read -r _id; do
        [[ -n "$_id" ]] && doomed["$_id"]=1
    done <<< "$remove_ids"

    # 1) fulls that a surviving differential still needs
    for _id in "${SNAP_IDS[@]}"; do
        [[ -n "${doomed[$_id]:-}" ]] && continue
        snap_has_tag "$_id" differential || continue
        _p=$(snap_tag_value "$_id" "parent-id=") || _p=""
        [[ -n "$_p" && -n "${doomed[$_p]:-}" ]] || continue
        unset "doomed[$_p]"
        reason["$_p"]="still referenced by a kept differential"
    done

    # 2) configs of everything that survives (including the fulls kept above)
    for _id in "${SNAP_IDS[@]}"; do
        [[ -n "${doomed[$_id]:-}" ]] && continue
        _ref=$(snap_tag_value "$_id" "config-id=") || _ref=""
        [[ -n "$_ref" && -n "${doomed[$_ref]:-}" ]] || continue
        unset "doomed[$_ref]"
        reason["$_ref"]="config of a kept backup"
    done

    local -a to_forget=()
    local _disk
    for _id in "${SNAP_IDS[@]}"; do
        [[ -n "${reason[$_id]:-}" ]] && {
            log_info "  Protected: ${_id}  ${SNAP_TIME[$_id]}  $(snap_kind "$_id")  $(snap_owner "$_id")  (${reason[$_id]})"
            continue
        }
        [[ -n "${doomed[$_id]:-}" ]] || continue
        _disk=$(snap_disk_tag "$_id") || _disk="-"
        log_info "  Prune: ${_id}  ${SNAP_TIME[$_id]}  $(snap_kind "$_id")  ${_disk:--}"
        to_forget+=("$_id")
    done

    if [[ ${#to_forget[@]} -gt 0 ]]; then
        if restic_logged "forget" forget "${to_forget[@]}"; then
            log_info "forget OK (${#to_forget[@]} removed, ${#reason[@]} protected)"
        else
            log_error "restic forget failed"
        fi
    else
        log_info "Nothing to forget (all ${#reason[@]} protected)"
    fi

    if [[ "$PRUNE_AFTER_BACKUP" == "true" ]]; then
        if restic_logged "prune" prune; then
            log_info "prune OK"
        else
            log_error "restic prune failed"
        fi
    fi
}

# ================== Backup overview ==================
# Identical to 'ResticRestore.sh --list --size': one line per snapshot with the
# cumulative restore size. Sizes cost one 'restic stats' call per snapshot, so
# a large repository takes a moment - set PRINT_BACKUP_OVERVIEW=false to skip.
#
# The SIZE column answers "what does restoring THIS line cost", not "what does
# it occupy in the repository":
#   FULL / RAW / LXC-FS / LOCAL : its own size
#   DIFF                        : itself plus its parent full
#   CONFIG                      : the whole run it belongs to
print_backup_overview() {
    [[ "$PRINT_BACKUP_OVERVIEW" == "true" ]] || return 0
    log_info "=== Backups in the repository ==="

    load_snapshot_index || { log_warn "Cannot read the snapshot list"; return 1; }
    if [[ ${#SNAP_IDS[@]} -eq 0 ]]; then
        log_info "Repository is empty"
        return 0
    fi

    local _id _kind _cfg _parent
    local -A KIND=() CFG=() PARENT=() SIZE=()
    for _id in "${SNAP_IDS[@]}"; do
        _kind=$(snap_kind "$_id")
        KIND["$_id"]="$_kind"
        if [[ "$_kind" == "CONFIG" ]]; then
            CFG["$_id"]="$_id"
        else
            _cfg=$(snap_tag_value "$_id" "config-id=") || _cfg=""
            CFG["$_id"]="$_cfg"
        fi
        _parent=$(snap_tag_value "$_id" "parent-id=") || _parent=""
        PARENT["$_id"]="$_parent"
        SIZE["$_id"]=$(snap_restore_size "$_id")
    done

    local rows="" _name _cum _k _pp
    for _id in "${SNAP_IDS[@]}"; do
        _cum=0
        case "${KIND[$_id]}" in
            DIFF)
                _cum=${SIZE[$_id]:-0}
                _pp="${PARENT[$_id]}"
                [[ -n "$_pp" ]] && _cum=$(( _cum + ${SIZE[$_pp]:-0} ))
                ;;
            CONFIG)
                # everything that carries the config-id of this run
                for _k in "${SNAP_IDS[@]}"; do
                    [[ "${CFG[$_k]}" == "$_id" && "$_k" != "$_id" ]] || continue
                    _cum=$(( _cum + ${SIZE[$_k]:-0} ))
                    if [[ "${KIND[$_k]}" == "DIFF" ]]; then
                        _pp="${PARENT[$_k]}"
                        [[ -n "$_pp" ]] && _cum=$(( _cum + ${SIZE[$_pp]:-0} ))
                    fi
                done
                ;;
            *)
                _cum=${SIZE[$_id]:-0}
                ;;
        esac
        _name=$(snap_tag_value "$_id" "name=") || _name=""
        rows+="${_id}"$'\t'"${SNAP_TIME[$_id]}"$'\t'"${KIND[$_id]}"$'\t'"$(snap_owner "$_id")"$'\t'"${_name:--}"$'\t'"$(snap_disk_column "$_id")"$'\t'"${PARENT[$_id]:--}"$'\t'"${CFG[$_id]:--}"$'\t'"$(human_bytes "$_cum")"$'\n'
    done

    local table
    table=$( { printf 'SNAP ID\tBackup Time\tTYPE\tMACHINE\tNAME\tDISK\tPARENT\tCONFIG\tSIZE\n'
               printf '%s' "$rows"
             } | awk -F'\t' '{printf "%-10s  %-16s  %-7s  %-9s  %-24s  %-10s  %10s  %-10s  %s\n", $1, $2, $3, $4, $5, $6, $9, $7, $8}' )
    local _line
    while IFS= read -r _line; do
        log_info "$_line"
    done <<< "$table"
    return 0
}

# ================== Inventory ==================
# Resolves and measures all volumes to be backed up ONCE per run. Without it
# the planning phase, the zfs send hint and the backup itself would repeat the
# very same 'pvesm path' and 'zfs get' calls.
#
# Line format:
#   INV_VM_DISKS / INV_CT_VOLS : id|disk_key|voltype|path|dataset|bytes
#   INV_LOCAL_DIRS             : path|tagname|bytes
# voltype : zvol | zfsdir | blockdev | file | dir
# bytes   : space in use, -1 = not determined here (directory trees: 'du' is
#           expensive and only the planning phase needs the value, and only
#           when no backup of that tag exists yet)
INV_VM_DISKS=()
INV_CT_VOLS=()
INV_LOCAL_DIRS=()
INV_TOTAL_USED=0
INV_BUILT=false


# Arguments: <vm ids...> -- <ct ids...>
build_inventory() {
    $INV_BUILT && return 0
    INV_BUILT=true
    INV_VM_DISKS=(); INV_CT_VOLS=(); INV_LOCAL_DIRS=(); INV_TOTAL_USED=0

    local -a vm_ids=() ct_ids=()
    local _cur="vm" _a
    for _a in "$@"; do
        if [[ "$_a" == "--" ]]; then _cur="ct"; continue; fi
        if [[ "$_cur" == "vm" ]]; then vm_ids+=("$_a"); else ct_ids+=("$_a"); fi
    done

    log_info "=== Inventory (volumes are resolved and measured once) ==="

    local vmid ctid CFG line key value volume _type _path _ds _sz
    for vmid in ${vm_ids[@]+"${vm_ids[@]}"}; do
        is_excluded "$vmid" && continue
        CFG=$(qm config "$vmid" 2>/dev/null) || continue
        if [[ "$SKIP_TEMPLATES" == true ]] && echo "$CFG" | grep -q '^template:.*1'; then
            continue
        fi
        while IFS= read -r line; do
            [[ -z "$line" ]] && continue
            key="${line%%:*}"
            value="${line#*:}"
            value="${value#"${value%%[![:space:]]*}"}"
            volume="${value%%,*}"
            volume="${volume#file=}"
            [[ -z "$volume" || "$volume" == "none" ]] && continue
            if [[ "$key" != efidisk* ]] && [[ ",${value}," == *",backup=0"* ]]; then
                continue
            fi
            # A CD/DVD drive is not a disk. Without this check an attached ISO
            # (ide2: local:iso/foo.iso,media=cdrom) resolves to a readable file
            # and would end up in the repository as a RAW image.
            if [[ ",${value}," == *",media=cdrom,"* ]]; then
                continue
            fi
            if ! resolve_volume "$volume"; then
                log_warn "  VM ${vmid} ${key}: cannot resolve volume ${volume} - skipping"
                continue
            fi
            _type="$VOL_KIND"; _path="$VOL_PATH"; _ds=""
            [[ "$_type" == "zvol" ]] && _ds="${_path#/dev/zvol/}"
            _sz=$(raw_allocated_bytes "$_path") || _sz=0
            [[ "$_sz" =~ ^[0-9]+$ ]] || _sz=0
            INV_VM_DISKS+=("${vmid}|${key}|${_type}|${_path}|${_ds}|${_sz}")
            INV_TOTAL_USED=$(( INV_TOTAL_USED + _sz ))
        done < <(echo "$CFG" | grep -E "$VM_DISK_REGEX" || true)
    done

    for ctid in ${ct_ids[@]+"${ct_ids[@]}"}; do
        is_excluded "$ctid" && continue
        CFG=$(pct config "$ctid" 2>/dev/null) || continue
        if [[ "$SKIP_TEMPLATES" == true ]] && echo "$CFG" | grep -q '^template:.*1'; then
            continue
        fi
        while IFS= read -r line; do
            [[ -z "$line" ]] && continue
            key="${line%%:*}"
            value="${line#*:}"
            value="${value#"${value%%[![:space:]]*}"}"
            volume="${value%%,*}"
            volume="${volume#file=}"
            [[ -z "$volume" || "$volume" == "none" ]] && continue
            if [[ ",${value}," == *",backup=0"* ]]; then
                continue
            fi
            if ! resolve_volume "$volume"; then
                log_warn "  CT ${ctid} ${key}: cannot resolve volume ${volume} - skipping"
                continue
            fi
            _type="$VOL_KIND"; _path="$VOL_PATH"; _ds=""
            if [[ "$_type" == "zvol" ]]; then
                _ds="${_path#/dev/zvol/}"
            elif [[ "$_type" == "dir" ]] && command -v zfs &>/dev/null; then
                _ds=$(zfs list -H -o name "$_path" 2>/dev/null) || _ds=""
                [[ -n "$_ds" ]] && _type="zfsdir"
            fi
            case "$_type" in
                zfsdir) _sz=$(zfs_used_bytes "$_ds") || _sz=0 ;;
                dir)    _sz=-1 ;;
                *)      _sz=$(raw_allocated_bytes "$_path") || _sz=0 ;;
            esac
            [[ "$_sz" =~ ^-?[0-9]+$ ]] || _sz=0
            INV_CT_VOLS+=("${ctid}|${key}|${_type}|${_path}|${_ds}|${_sz}")
            [[ "$_sz" -gt 0 ]] && INV_TOTAL_USED=$(( INV_TOTAL_USED + _sz ))
        done < <(echo "$CFG" | grep -E "$LXC_FS_REGEX" || true)
    done

    if [[ -n "$LOCAL_DIRS" ]]; then
        local _oldifs="$IFS" _dir _dirname
        IFS=$' \t\n'
        for _dir in $LOCAL_DIRS; do
            _dir="${_dir%\"}"
            _dir="${_dir#\"}"
            [[ -d "$_dir" ]] || continue
            _dirname="${_dir%/}"
            _dirname="${_dirname##*/}"
            INV_LOCAL_DIRS+=("${_dir}|${_dirname}|-1")
        done
        IFS="$_oldifs"
    fi

    log_info "  ${#INV_VM_DISKS[@]} VM disk(s), ${#INV_CT_VOLS[@]} CT volume(s), ${#INV_LOCAL_DIRS[@]} local dir(s) | in use: $(human_bytes "$INV_TOTAL_USED")"
    return 0
}

# Print all inventory rows of one machine.
inv_entries() {
    local kind="$1" id="$2" e
    if [[ "$kind" == "vm" ]]; then
        for e in ${INV_VM_DISKS[@]+"${INV_VM_DISKS[@]}"}; do
            [[ "${e%%|*}" == "$id" ]] && echo "$e"
        done
    else
        for e in ${INV_CT_VOLS[@]+"${INV_CT_VOLS[@]}"}; do
            [[ "${e%%|*}" == "$id" ]] && echo "$e"
        done
    fi
    return 0
}


# ================== zfs send selection ==================
# BACKUP_MODE=zfs backs every zvol up in the classic zfs send format.
#
# In raw mode zfs send is the exception, chosen per DISK rather than per machine:
# it writes a full generation every FULL_MAX_AGE_DAYS days, which is what makes
# it expensive on a tight target drive. That price is worth paying for a large
# data disk that barely changes, and not for the OS disk of the same machine.

# Does the operator want zfs send for this disk? $1 vmid, $2 disk key.
disk_wants_zfs_send() {
    [[ "$BACKUP_MODE" == "zfs" ]] && return 0
    [[ ",${ZFS_SEND_DISKS}," == *",${1}:${2},"* ]]
}

# ... and can this volume actually deliver it? Only ZFS zvols can.
# $1 vmid, $2 disk key, $3 volume type.
volume_uses_zfs_send() {
    [[ "$3" == "zvol" ]] || return 1
    disk_wants_zfs_send "$1" "$2"
}

# RAW re-reads the whole disk on every run; 'zfs send -i' reads only what
# changed. On a disk that is large and barely changes, that read effort is the
# one remaining reason to prefer zfs send - space is not, because restic
# deduplicates the unchanged blocks of a RAW image away anyway.
# The churn is taken from ZFS metadata (written@, between the two newest
# snapshots), so this costs no data reads. Reported per disk, because that is
# the granularity ZFS_SEND_DISKS works at.
zfs_send_hint() {
    [[ "$ZFS_SEND_HINT" == "true" ]] || return 0
    command -v zfs &>/dev/null || return 0
    [[ ${#INV_VM_DISKS[@]} -gt 0 ]] || return 0

    local min_bytes=$(( ZFS_SEND_HINT_MIN_GB * 1024 * 1024 * 1024 ))
    local entry vmid key vtype vpath vds vsz written pct h suggest
    local -a snaps=() hints=() descriptors=()

    for entry in "${INV_VM_DISKS[@]}"; do
        IFS='|' read -r vmid key vtype vpath vds vsz <<< "$entry"
        [[ "$vtype" == "zvol" ]] || continue
        disk_wants_zfs_send "$vmid" "$key" && continue
        is_always_full_disk "$key" && continue
        [[ "$vsz" -ge "$min_bytes" ]] || continue

        # the two newest snapshots bracket one backup interval
        mapfile -t snaps < <(zfs list -t snapshot -H -o name -S creation "$vds" 2>/dev/null | head -2 || true)
        [[ ${#snaps[@]} -eq 2 ]] || continue
        written=$(zfs get -Hp -o value "written@${snaps[1]#*@}" "${snaps[0]}" 2>/dev/null) || continue
        [[ "$written" =~ ^[0-9]+$ ]] || continue
        pct=$(( written * 100 / vsz ))
        [[ "$pct" -le "$ZFS_SEND_HINT_MAX_CHURN_PERCENT" ]] || continue
        hints+=("  ${vmid}:${key} - $(human_bytes "$vsz"), ${pct}% of it changed since the previous snapshot")
        descriptors+=("${vmid}:${key}")
    done

    [[ ${#hints[@]} -gt 0 ]] || return 0
    log_info "=== zfs send hint ==="
    for h in "${hints[@]}"; do log_info "$h"; done
    log_info "  Large and cold: a RAW image reads these disks completely on every"
    log_info "  run, 'zfs send -i' would only read the changed blocks. If the"
    log_info "  target drive has room for a second full generation of them:"
    suggest=$(IFS=,; echo "${descriptors[*]}")
    if [[ -n "$ZFS_SEND_DISKS" ]]; then
        # Deliberately not the merged list: it would echo back any typo already
        # in the setting and invite pasting it in again.
        log_info "      add to ZFS_SEND_DISKS: ${suggest}"
    else
        log_info "      ZFS_SEND_DISKS=\"${suggest}\""
    fi
    return 0
}

# ================== Planning phase ==================# ================== Planning phase ==================
# Determines, WITHOUT writing any data, per disk whether a full or a
# differential is due and what the run needs in bytes (REQUIRED_BYTES) for the
# space check. The only source is the inventory - nothing is resolved here.

# "vmid|disk_key|dataset|current_snap|base_snap"  (base empty -> full needed)
PLAN_ZFS_DISKS=()
# "prefix|id|disk_key|source_path|bytes"
PLAN_RAW_VOLUMES=()


# A descriptor pointing at a disk that does not exist matches nothing and would
# silently keep costing a full read on every run. The form was validated at
# startup; this checks it against the disks actually found.
# Skipped on a partial run (specific VMIDs on the command line), where
# descriptors for the other machines are absent for a legitimate reason.
check_zfs_send_disks() {
    [[ -n "$ZFS_SEND_DISKS" ]] || return 0
    $RB_PARTIAL_RUN && return 0

    local -A present=()
    local entry vmid key rest e
    local -a wanted=()
    for entry in ${INV_VM_DISKS[@]+"${INV_VM_DISKS[@]}"}; do
        IFS='|' read -r vmid key rest <<< "$entry"
        present["${vmid}:${key}"]=1
    done
    IFS=',' read -ra wanted <<< "$ZFS_SEND_DISKS"
    for e in ${wanted[@]+"${wanted[@]}"}; do
        [[ -n "$e" ]] || continue
        [[ -n "${present[$e]:-}" ]] && continue
        log_warn "  ZFS_SEND_DISKS entry '${e}' matches no disk in this run - typo, or the machine is excluded or gone"
    done
    return 0
}

plan_all_backups() {
    PLAN_ZFS_DISKS=()
    PLAN_RAW_VOLUMES=()
    SUM_ZFS_BYTES=0
    SUM_RAW_BYTES=0
    SUM_OTHER_BYTES=0
    REQUIRED_BYTES=0

    log_info "=== Planning phase (full/diff decision) ==="
    log_info "  BACKUP_MODE=${BACKUP_MODE}${ZFS_SEND_DISKS:+   ZFS_SEND_DISKS=${ZFS_SEND_DISKS}}"

    local entry vmid key vtype vpath vds vsz disk_tag cur_snap base_snap
    for entry in ${INV_VM_DISKS[@]+"${INV_VM_DISKS[@]}"}; do
        IFS='|' read -r vmid key vtype vpath vds vsz <<< "$entry"

        if ! volume_uses_zfs_send "$vmid" "$key" "$vtype"; then
            if disk_wants_zfs_send "$vmid" "$key"; then
                log_warn "  ${vmid}:${key} - zfs send requested, but this is not a ZFS zvol (${vtype}) - storing a RAW image"
            fi
            PLAN_RAW_VOLUMES+=("vm|${vmid}|${key}|${vpath}|${vsz}")
            continue
        fi

        disk_tag="vm-${vmid}-${key}"
        base_snap=""
        if ! find_latest_snap "$vds"; then
            log_info "  VM ${vmid} ${key}: no snapshot -> full required"
            PLAN_ZFS_DISKS+=("${vmid}|${key}|${vds}||")
            continue
        fi
        cur_snap="${LATEST_SNAP#*@}"
        if is_always_full_disk "$key"; then
            log_info "  VM ${vmid} ${key}: always backed up full (EFI/TPM state)"
        elif disk_needs_full "$disk_tag" "$vds"; then
            log_info "  VM ${vmid} ${key}: full required (${DISK_FULL_REASON})"
        else
            base_snap="$BASE_SNAP_NAME"
        fi
        PLAN_ZFS_DISKS+=("${vmid}|${key}|${vds}|${cur_snap}|${base_snap}")
    done

    check_zfs_send_disks

    if [[ ${#PLAN_ZFS_DISKS[@]} -eq 0 ]]; then
        log_info "  Result: every volume is stored as a RAW image, no differentials"
    else
        log_info "  Result: ${#PLAN_ZFS_DISKS[@]} disk(s) via zfs send, ${#PLAN_RAW_VOLUMES[@]} as RAW image(s)"
    fi
    return 0
}

# Plan LXC containers: the requirement depends on the storage type.
plan_ct_backups() {
    local ctid key vtype vpath vds vsz entry need
    for entry in ${INV_CT_VOLS[@]+"${INV_CT_VOLS[@]}"}; do
        IFS='|' read -r ctid key vtype vpath vds vsz <<< "$entry"
        case "$vtype" in
            zfsdir)
                # Directory dataset: always file based, RAW makes no sense here
                need=$(estimate_backup_need "ct-${ctid},lxc" echo "$vsz")
                SUM_OTHER_BYTES=$(( SUM_OTHER_BYTES + need ))
                log_info "  CT ${ctid} ${key}: requires ~$(human_bytes "$need") (ZFS, file-based)"
                ;;
            dir)
                if [[ "$vsz" -ge 0 ]]; then
                    need=$(estimate_backup_need "ct-${ctid},lxc" echo "$vsz")
                else
                    need=$(estimate_backup_need "ct-${ctid},lxc" dir_size_bytes "$vpath")
                fi
                SUM_OTHER_BYTES=$(( SUM_OTHER_BYTES + need ))
                log_info "  CT ${ctid} ${key}: requires ~$(human_bytes "$need") (dir, file-based)"
                ;;
            *)
                PLAN_RAW_VOLUMES+=("ct|${ctid}|${key}|${vpath}|${vsz}")
                ;;
        esac
    done
    return 0
}

# Plan the LOCAL_DIRS backups.
plan_local_dirs() {
    local entry _dir _dirname _sz need
    for entry in ${INV_LOCAL_DIRS[@]+"${INV_LOCAL_DIRS[@]}"}; do
        IFS='|' read -r _dir _dirname _sz <<< "$entry"
        if [[ "$_sz" -ge 0 ]]; then
            need=$(estimate_backup_need "local,dir-${_dirname}" echo "$_sz")
        else
            need=$(estimate_backup_need "local,dir-${_dirname}" dir_size_bytes "$_dir")
        fi
        SUM_OTHER_BYTES=$(( SUM_OTHER_BYTES + need ))
        log_info "  LOCAL ${_dir}: requires ~$(human_bytes "$need")"
    done
    return 0
}

# Determine sizes and build REQUIRED_BYTES. Only here do the expensive
# estimates run - and only those needed for the decided run type.
# Determine sizes and build REQUIRED_BYTES. Only here do the expensive
# estimates run - and only those the planned run actually needs.
finalize_space_plan() {
    local entry _vmid _key _ds _cur _base _sz
    local rv _pfx _id _rkey _path _psz _need

    for entry in ${PLAN_ZFS_DISKS[@]+"${PLAN_ZFS_DISKS[@]}"}; do
        IFS='|' read -r _vmid _key _ds _cur _base <<< "$entry"
        if [[ -z "$_base" ]]; then
            _sz=$(zfs_used_bytes "$_ds") || _sz=0
            log_info "  VM ${_vmid} ${_key}: full ~$(human_bytes "$_sz")"
        elif [[ "$_base" == "$_cur" ]]; then
            # snapshot unchanged since the last full -> the disk will be skipped
            _sz=0
            log_info "  VM ${_vmid} ${_key}: unchanged since the last full - nothing to write"
        else
            _sz=$(zfs_send_size "$_ds" "$_cur" "$_base") || {
                log_warn "  VM ${_vmid} ${_key}: cannot determine diff size - falling back to ZFS used"
                _sz=$(zfs_used_bytes "$_ds") || _sz=0
            }
            log_info "  VM ${_vmid} ${_key}: diff ~$(human_bytes "$_sz") (${_base} -> ${_cur})"
        fi
        SUM_ZFS_BYTES=$(( SUM_ZFS_BYTES + _sz ))
    done

    for rv in ${PLAN_RAW_VOLUMES[@]+"${PLAN_RAW_VOLUMES[@]}"}; do
        IFS='|' read -r _pfx _id _rkey _path _psz <<< "$rv"
        if [[ "$_psz" =~ ^[0-9]+$ ]]; then
            _need=$(estimate_backup_need "${_pfx}-${_id}-${_rkey},raw" echo "$_psz")
        else
            _need=$(estimate_backup_need "${_pfx}-${_id}-${_rkey},raw" raw_allocated_bytes "$_path")
        fi
        SUM_RAW_BYTES=$(( SUM_RAW_BYTES + _need ))
        log_info "  ${_pfx}-${_id} ${_rkey}: RAW requires ~$(human_bytes "$_need")"
    done

    REQUIRED_BYTES=$(( SUM_ZFS_BYTES + SUM_RAW_BYTES + SUM_OTHER_BYTES ))

    log_info "=== Space required for this run ==="
    if [[ ${#PLAN_ZFS_DISKS[@]} -gt 0 ]]; then
        log_info "  VM disks (zfs send): $(human_bytes "$SUM_ZFS_BYTES")"
    fi
    if [[ ${#PLAN_RAW_VOLUMES[@]} -gt 0 ]]; then
        log_info "  RAW images:          $(human_bytes "$SUM_RAW_BYTES")  (always full)"
    fi
    if [[ "$SUM_OTHER_BYTES" -gt 0 ]]; then
        log_info "  LXC/LOCAL_DIRS:      $(human_bytes "$SUM_OTHER_BYTES")  (new = full size, existing = ${ESTIMATED_BACKUP_INCREASE_PERCENTAGE}%)"
    fi
    log_info "  Total required:      $(human_bytes "$REQUIRED_BYTES")"
    return 0
}

# ================== Main ResticBackup Routine (called after mount) ==================
run_restic_backup() {
    local RB_VMIDS_TO_BACKUP=()
    local arg
    for arg in "$@"; do
        case "$arg" in
            --dry-run)    RESTIC_DRY_RUN=true ;;
            --verbose|-v) RESTIC_VERBOSE=true ;;
            *)
                if [[ "$arg" =~ ^[0-9]+$ ]]; then
                    RB_VMIDS_TO_BACKUP+=("$arg")
                fi
                ;;
        esac
    done

    [[ ${#RB_VMIDS_TO_BACKUP[@]} -gt 0 ]] && RB_PARTIAL_RUN=true

    normalize_compression RESTIC_COMPRESSION_SEND
    normalize_compression RESTIC_COMPRESSION_RAW
    normalize_compression RESTIC_COMPRESSION_FILES

    LXC_EXCLUDE_FLAGS=()
    local _dirs
    IFS=',' read -ra _dirs <<< "$LXC_EXCLUDE_DIRS"
    local _d
    for _d in "${_dirs[@]}"; do
        _d="${_d#"${_d%%[![:space:]]*}"}"
        _d="${_d%"${_d##*[![:space:]]}"}"
        _d="${_d#/}"
        [[ -n "$_d" ]] && LXC_EXCLUDE_FLAGS+=(--exclude "**/${_d}")
    done

    # Excludes for LOCAL_DIRS backups (default: the .zfs snapdir)
    LOCAL_EXCLUDE_FLAGS=()
    local _ldirs _ld
    IFS=',' read -ra _ldirs <<< "$LOCAL_EXCLUDE_DIRS"
    for _ld in "${_ldirs[@]}"; do
        _ld="${_ld#"${_ld%%[![:space:]]*}"}"
        _ld="${_ld%"${_ld##*[![:space:]]}"}"
        _ld="${_ld#/}"
        [[ -n "$_ld" ]] && LOCAL_EXCLUDE_FLAGS+=(--exclude "**/${_ld}")
    done

    if [[ ! -f "$RESTIC_PASSWORDFILE" ]]; then
        log_info "Password file not found, auto-creating: ${RESTIC_PASSWORDFILE}"
        local _pw
        _pw=$(openssl rand -base64 32)
        echo -n "$_pw" > "$RESTIC_PASSWORDFILE"
        chmod 600 "$RESTIC_PASSWORDFILE"
        log_info "Password generated (32 chars, saved with chmod 600)"
        log_warn "IMPORTANT: Save this password elsewhere, it is needed for restore!"
        log_warn "  cat '${RESTIC_PASSWORDFILE}'"
    fi

    log_info "Base: ${RESTIC_BASEFOLDER}"
    log_info "Repo: ${RESTIC_REPOSITORY}"

    # Must run BEFORE the repo check below: a leftover lock makes
    # 'restic snapshots' fail, which would look like an uninitialized repo and
    # send us into 'restic init' on an existing repository.
    clear_stale_repo_lock

    local _snap_opts=()
    $RESTIC_DRY_RUN && _snap_opts+=(--no-lock)
    if ! restic snapshots "${_snap_opts[@]}" &>/dev/null; then
        if $RESTIC_DRY_RUN; then
            log_warn "Restic repo not initialized (dry-run, would auto-create)"
        else
            log_info "Restic repo not found, auto-initializing: ${RESTIC_REPOSITORY}"
            restic_logged "init" init
            log_info "Restic repo initialized and ready"
        fi
    fi

    log_info "=== Starting Proxmox backup ==="
    $RESTIC_DRY_RUN && log_info "DRY-RUN MODE"

    VM_DISK_REGEX='^(ide[0-3]|sata[0-5]|scsi[0-9]+|virtio[0-9]+|efidisk[0-9]+|tpmstate[0-9]+):'
    LXC_FS_REGEX='^rootfs:'

    local VM_IDS=() CT_IDS=()
    if [[ ${#RB_VMIDS_TO_BACKUP[@]} -gt 0 ]]; then
        local _id
        for _id in "${RB_VMIDS_TO_BACKUP[@]}"; do
            if qm config "$_id" &>/dev/null; then
                VM_IDS+=("$_id")
            fi
        done
    else
        mapfile -t VM_IDS < <(qm list 2>/dev/null | awk 'NR > 1 && NF { print $1 }')
        [[ ${#VM_IDS[@]} -eq 0 ]] && log_info "No VMs found"
    fi

    if [[ ${#RB_VMIDS_TO_BACKUP[@]} -eq 0 ]]; then
        mapfile -t CT_IDS < <(pct list 2>/dev/null | awk 'NR>1 && NF {print $1}')
        [[ ${#CT_IDS[@]} -eq 0 ]] && log_info "No LXC containers found"
    else
        local id
        for id in "${RB_VMIDS_TO_BACKUP[@]}"; do
            if pct config "$id" &>/dev/null; then
                CT_IDS+=("$id")
            fi
        done
    fi

    if [[ ${#RB_VMIDS_TO_BACKUP[@]} -gt 0 ]]; then
        local _id2 _is_vm _is_ct _v _c
        for _id2 in "${RB_VMIDS_TO_BACKUP[@]}"; do
            _is_vm=false; _is_ct=false
            for _v in "${VM_IDS[@]}"; do [[ "$_v" == "$_id2" ]] && _is_vm=true && break; done
            for _c in "${CT_IDS[@]}"; do [[ "$_c" == "$_id2" ]] && _is_ct=true && break; done
            if ! $_is_vm && ! $_is_ct; then
                log_warn "VMID ${_id2} is neither a VM nor an LXC container - ignoring"
            fi
        done
    fi

    # --- Inventory: resolve and measure all volumes once ---
    build_inventory ${VM_IDS[@]+"${VM_IDS[@]}"} -- ${CT_IDS[@]+"${CT_IDS[@]}"}

    # --- Planning phase: decide full/diff + determine the exact space needed ---
    plan_all_backups
    plan_ct_backups
    plan_local_dirs
    finalize_space_plan

    # --- Space check / release before the backup ---
    ensure_space_available

    # --- Is a volume worth switching to zfs send? ---
    zfs_send_hint

    # Measured AFTER the pre-backup pruning on purpose. Taken before, a pruning
    # run would make the repository smaller than the reference value, the delta
    # below would go negative, get clamped to 0 - and run_post_backup_check
    # would silently skip the integrity check with "no new data written".
    if ! $RESTIC_DRY_RUN; then
        REPO_SIZE_BEFORE=$(restic stats --mode raw-data --json 2>/dev/null \
            | grep -x '^{.*}' | jq -r '.total_size // 0' 2>/dev/null) || REPO_SIZE_BEFORE=0
        log_info "Repo data size before backup: $(human_bytes "$REPO_SIZE_BEFORE")"
    fi

    local vmid
    for vmid in "${VM_IDS[@]}"; do
        if is_excluded "$vmid"; then
            log_info "Skip VM ${vmid} (excluded)"
            TOTAL_SKIPPED=$((TOTAL_SKIPPED+1))
            continue
        fi
        local VM_CONFIG VM_NAME
        VM_CONFIG=$(qm config "$vmid" 2>/dev/null) || { log_error "qm config ${vmid} failed"; continue; }
        VM_NAME=$(echo "$VM_CONFIG" | awk -F': ' '/^name:/{print $2}')
        NAME_TAG_FLAGS=()
        [[ -n "$VM_NAME" ]] && NAME_TAG_FLAGS=(--tag "name=$(sanitize_tag_value "$VM_NAME")")
        if [[ "$SKIP_TEMPLATES" == true ]] && echo "$VM_CONFIG" | grep -q '^template:.*1'; then
            log_info "Skip VM ${vmid} (${VM_NAME}) - template"
            TOTAL_SKIPPED=$((TOTAL_SKIPPED+1))
            continue
        fi
        log_info "--- VM ${vmid} (${VM_NAME}) ---"
        TOTAL_VMS=$((TOTAL_VMS+1))

        # Back up the config first -> its short ID groups this backup run
        backup_config "$vmid" "qemu-server" "vm" || true
        CONFIG_ID_TAG_FLAGS=()
        [[ -n "$CURRENT_CONFIG_ID" ]] && CONFIG_ID_TAG_FLAGS=(--tag "config-id=${CURRENT_CONFIG_ID}")

        local _iid key vtype vpath vds vsz
        while IFS='|' read -r _iid key vtype vpath vds vsz; do
            [[ -z "$key" ]] && continue
            if volume_uses_zfs_send "$vmid" "$key" "$vtype"; then
                backup_zfs_disk "$vmid" "$key" "$vds" || true
            else
                backup_raw_volume "vm" "$vmid" "$key" "$vpath" || true
            fi
        done < <(inv_entries vm "$vmid")
    done

    local ctid
    for ctid in "${CT_IDS[@]}"; do
        if is_excluded "$ctid"; then
            log_info "Skip CT ${ctid} (excluded)"
            TOTAL_SKIPPED=$((TOTAL_SKIPPED+1))
            continue
        fi
        local CT_CONFIG CT_NAME
        CT_CONFIG=$(pct config "$ctid" 2>/dev/null) || { log_error "pct config ${ctid} failed"; continue; }
        CT_NAME=$(echo "$CT_CONFIG" | awk '/^hostname:/{print $2}')
        [[ -z "$CT_NAME" ]] && CT_NAME="(unnamed)"
        NAME_TAG_FLAGS=(--tag "name=$(sanitize_tag_value "$CT_NAME")")
        if [[ "$SKIP_TEMPLATES" == true ]] && echo "$CT_CONFIG" | grep -q '^template:.*1'; then
            log_info "Skip CT ${ctid} (${CT_NAME}) - template"
            TOTAL_SKIPPED=$((TOTAL_SKIPPED+1))
            continue
        fi
        log_info "--- CT ${ctid} (${CT_NAME}) ---"
        TOTAL_CTS=$((TOTAL_CTS+1))

        # Back up the config first -> its short ID groups this backup run
        backup_config "$ctid" "lxc" "ct" || true
        CONFIG_ID_TAG_FLAGS=()
        [[ -n "$CURRENT_CONFIG_ID" ]] && CONFIG_ID_TAG_FLAGS=(--tag "config-id=${CURRENT_CONFIG_ID}")

        local _iid key vtype vpath vds vsz
        while IFS='|' read -r _iid key vtype vpath vds vsz; do
            [[ -z "$key" ]] && continue
            case "$vtype" in
                zfsdir)
                    # Directory dataset: always file based via a ro snapshot
                    backup_lxc_fs "$ctid" "$key" "$vds" || true ;;
                dir)
                    backup_lxc_dir "$ctid" "$key" "$vpath" || true ;;
                zvol|blockdev|file)
                    backup_raw_volume "ct" "$ctid" "$key" "$vpath" || true ;;
                *)
                    log_warn "  ${key}: volume type '${vtype}' not supported (${vpath})" ;;
            esac
        done < <(inv_entries ct "$ctid")
    done

    if [[ -n "$LOCAL_DIRS" ]]; then
        NAME_TAG_FLAGS=()
        CONFIG_ID_TAG_FLAGS=()
        log_info "=== Local directories ==="
        local _entry _ldir _lname _lsz
        for _entry in ${INV_LOCAL_DIRS[@]+"${INV_LOCAL_DIRS[@]}"}; do
            IFS='|' read -r _ldir _lname _lsz <<< "$_entry"
            backup_local_dir "$_ldir" || true
        done
    fi

    if ! $RESTIC_DRY_RUN; then
        local _repo_after
        _repo_after=$(restic stats --mode raw-data --json 2>/dev/null \
            | grep -x '^{.*}' | jq -r '.total_size // 0' 2>/dev/null) || _repo_after=0
        TOTAL_DATA_ADDED=$(( _repo_after - REPO_SIZE_BEFORE ))
        [[ "$TOTAL_DATA_ADDED" -lt 0 ]] && TOTAL_DATA_ADDED=0
        log_info "Data added this run: $(human_bytes "$TOTAL_DATA_ADDED")"
    fi

    if ! $RESTIC_DRY_RUN; then
        check_disk_space
    fi

    log_info "=== Proxmox backup complete ==="
    log_info "VMs=${TOTAL_VMS} CTs=${TOTAL_CTS} Disks=${TOTAL_DISKS} (Full=${TOTAL_FULLS} Diff=${TOTAL_DIFFS} Raw=${TOTAL_RAWS}) Skipped=${TOTAL_SKIPPED} | Written=$(human_bytes "$TOTAL_DATA_ADDED") | Errors=${ERRORS} Warnings=${WARNINGS}"

    if ! $RESTIC_DRY_RUN; then
        run_forget_prune
        run_post_backup_check
        print_backup_overview
    fi

        return $((ERRORS > 0 ? 1 : 0))
}

fs_trim_now () {
    if [[ "${fsTrim}" == "yes" ]]; then
        echo "fstrim ${RESTIC_BASEFOLDER}"
        fstrim -v "$RESTIC_BASEFOLDER" || echo "fstrim not supported"
    else
        echo "fstrim not requested"
    fi
}

check_medium_change () {
    if [[ "${COUNTIGNORE}" == "no" ]]; then
        echo "========================"
        echo "Check Backup Media Count:"
        local BD BC
        BD=$(tail -n 30 "${logPath}" 2>/dev/null \
            | awk '{for(i=1;i<=NF;i++) if($i ~ /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/) print $i}' \
            | sort -u || true)
        BC=$(echo "${BD}" | grep -c . || true)
        echo "${BD}"
        if [[ "${BC}" -le 1 ]] ; then
            echo "WARNING: backup medium has not been rotated for too long! Only ${BC} medium/media found!"
        else
            echo "${BC} different backup media found."
        fi
    fi
}

check_filesystem_usage() {
    local usage_all usage_full
    echo "========================"
    echo "Check File System Usage:"
    usage_all=$(df -h | awk '!/efivars/')
    usage_full=$(echo "${usage_all}" \
        | awk -v limit="${FULL_FILESYSTEM_PERCENTAGE}" '($5+0) >= limit')
    if [[ -n "${usage_full}" ]]; then
        echo "${usage_full}" | sed 's/^/WARNING: /'
    fi
    echo "${usage_all}"
}

# Warn if the uptime exceeds MAX_UPTIME_DAYS (updates probably not applied).
check_uptime() {
    local uptime_days

    echo "========================"
    echo "Check Uptime / Update Status:..."

    uptime_days=$(( $(awk '{print int($1)}' /proc/uptime) / 86400 ))

    if [[ "${uptime_days}" -gt "${MAX_UPTIME_DAYS}" ]]; then
        echo "WARNING: uptime is ${uptime_days} days: system has been running too long, updates probably not applied!"
    else
        echo "Uptime is ${uptime_days} days"
    fi
}

check_pool_status() {
    local degraded

    echo "========================"
    echo "Check Pool Status:"

    if ! command -v zpool &>/dev/null; then
        echo "No ZFS on this host - pool status skipped"
        return 0
    fi

    zpool status
    degraded=$(zpool status 2>/dev/null | grep degraded || true)

    if [[ -n "${degraded}" ]]; then
        echo "WARNING: ${degraded}"
    fi
}

check_medium_change
check_filesystem_usage
check_uptime
check_pool_status


mkdir -p "${RESTIC_BASEFOLDER}"

echo "========================"
if ! mountpoint -q "${RESTIC_BASEFOLDER}"; then
   for uuid in "${UUIDS[@]}"
   do
      mount "UUID=$uuid" "${RESTIC_BASEFOLDER}" "${MOUNTOPT[@]}" || true
      mountpoint -q "${RESTIC_BASEFOLDER}" || continue
      echo "Backing up to UUID $uuid"
      echo "Backup started at: $(date +"%Y-%m-%d-%H:%M") on UUID $uuid" >> "${logPath}"
      dev=$(blkid -U "$uuid")
      break
   done
   if ! mountpoint -q "${RESTIC_BASEFOLDER}"; then
      echo "WARNING: no backup medium found"
      exit 1
   fi
else
   echo "Backing up to existing mountpoint"
   dev=$(findmnt -n -o SOURCE --target "$RESTIC_BASEFOLDER")
   echo "Backup started at: $(date +"%Y-%m-%d-%H:%M")" >> "${logPath}"
fi

# Do Backup
run_restic_backup "$@" || echo "WARNING: ResticBackup reported errors (see log above)"

if command -v mail &>/dev/null; then
    _send_notify=false
    SUBJECT="[Proxmox] Backup - "
    if [[ "$ERRORS" -gt 0 ]]; then
        SUBJECT+="FAILED (${ERRORS} errors)"
        _send_notify=true
    elif [[ "$NOTIFY_ON_SUCCESS" == "true" && -n "${LOG_OUTPUT}" ]]; then
        SUBJECT+="SUCCESS"
        _send_notify=true
    fi
    if $_send_notify; then
        printf '%b' "$LOG_OUTPUT" | mail -s "$SUBJECT" "$NOTIFY_EMAIL" 2>/dev/null || true
    fi
fi

fs_trim_now

if [[ "${mountPersistent}" == "no" ]]; then
   echo "Unmounting ${RESTIC_BASEFOLDER}"
   # Without the guard a busy mountpoint would trip the ERR trap and abort the
   # script right here - the disk would never be sent to standby.
   umount "${RESTIC_BASEFOLDER}" || echo "WARNING: unmounting ${RESTIC_BASEFOLDER} failed (still in use?)"
else
   echo "Unmount not requested: ${RESTIC_BASEFOLDER}"
fi

if [[ "${setStandby}" == "yes" ]]; then
   disk=""
   if [[ -n "${dev:-}" ]]; then
      # $dev can be a partition (/dev/sda1) or already the whole disk
      # (/dev/sda). PKNAME only returns the parent for partitions; for a whole
      # disk the result is empty and $dev is already the target for hdparm.
      pkname=$(lsblk -no PKNAME "${dev}" 2>/dev/null | awk 'NF{print;exit}')
      if [[ -n "$pkname" ]]; then
         disk="/dev/${pkname}"
      elif [[ -b "${dev}" ]]; then
         disk="${dev}"
      fi
   fi
   if [[ -n "$disk" && -b "$disk" ]]; then
      echo "Setting standby: $disk"
      hdparm -y "$disk" || echo "WARNING: standby failed"
   else
      echo "WARNING: cannot determine physical device for standby (dev='${dev:-}') - standby skipped"
   fi
else
   echo "Disk standby not requested"
fi