#!/usr/bin/env bash
set -euo pipefail

TARGET_MOUNT="${TARGET_MOUNT:-/userdata}"
DATA_ROOT="${DATA_ROOT:-/userdata/data}"
CONTAINER="smart-cam"
PROCESSED_FILE="/smart-cam/kafka_sender/processed_files.log"
SCAN_PROCESSED_FILE="${DATA_ROOT}/upload_processed_scans.log"
SCAN_RECEIPTS_FILE="${DATA_ROOT}/upload_completed_scans.json"
STATE_DIR="${STATE_DIR:-/var/lib/cleanup-smartcam-data}"
RECORDING_PAUSE_MARKER="${RECORDING_PAUSE_MARKER:-${DATA_ROOT}/.recording-paused-low-disk}"
LEGACY_PAUSED_STATE="${STATE_DIR}/capture-paused"
LEGACY_OLD_PAUSED_STATE="/var/lib/userdata-video-cleanup/capture-paused"
LOCK_FILE="${LOCK_FILE:-/run/cleanup-smartcam-data.lock}"

# Delete confirmed bundles on every scheduled run. Disable only local video
# recording below 3 GiB; detection, RTSP/live output and upload stay running.
# Re-enable recording after recovery to 5 GiB.
RESUME_TARGET_BYTES="${RESUME_TARGET_BYTES:-5368709120}"
PAUSE_BELOW_BYTES="${PAUSE_BELOW_BYTES:-3221225472}"
CRITICAL_BELOW_BYTES="${CRITICAL_BELOW_BYTES:-3221225472}"
MIN_FILE_AGE_MINUTES="${MIN_FILE_AGE_MINUTES:-60}"
SAFE_WINDOW_SECONDS="${SAFE_WINDOW_SECONDS:-10}"

mkdir -p "$STATE_DIR"
exec 9>"$LOCK_FILE"
flock -n 9 || exit 0

log() {
  local level="$1"
  shift
  local message="[DISK_GUARD][${level}] $*"
  printf '%s\n' "$message"
  logger -t cleanup-smartcam-data -- "$message" || true
}

container_log() {
  local message="$1"
  if docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null | grep -qx true; then
    docker exec "$CONTAINER" sh -c \
      'printf "%s\n" "$1" > /proc/1/fd/1' sh "$message" 2>/dev/null || true
  fi
}

alert() {
  local level="$1"
  shift
  local message="[DISK_GUARD][${level}] $*"
  log "$level" "$*"
  container_log "$message"
}

available_bytes() {
  df -PB1 "$TARGET_MOUNT" | awk 'NR == 2 {print $4}'
}

human_bytes() {
  numfmt --to=iec-i --suffix=B "$1"
}

camera_host_pid() {
  docker top "$CONTAINER" -eo pid,args 2>/dev/null |
    awk '$0 ~ /build\/bin\/smart_camera([[:space:]]|$)/ {print $1; exit}'
}

valid_camera_pid() {
  local pid="$1"
  [[ "$pid" =~ ^[0-9]+$ ]] &&
    [[ -r "/proc/${pid}/cmdline" ]] &&
    tr '\0' ' ' < "/proc/${pid}/cmdline" | grep -q 'build/bin/smart_camera'
}

camera_is_stopped() {
  local pid="$1"
  ps -o stat= -p "$pid" 2>/dev/null | grep -q '^T'
}

start_fast_resume_checks() {
  systemctl start cleanup-smartcam-data-resume.timer 2>/dev/null ||
    log ERROR "failed to start fast resume checks"
}

stop_fast_resume_checks() {
  systemctl --no-block stop cleanup-smartcam-data-resume.timer 2>/dev/null || true
}

recover_legacy_capture_pause() {
  local state_file pid
  for state_file in "$LEGACY_PAUSED_STATE" "$LEGACY_OLD_PAUSED_STATE"; do
    [[ -f "$state_file" ]] || continue
    pid="$(sed -n 's/^pid=//p' "$state_file" | head -1)"
    if valid_camera_pid "$pid" && camera_is_stopped "$pid"; then
      kill -CONT "$pid"
      alert RECOVERED "released legacy whole-process pause; detection and live stream remain running"
    fi
    rm -f "$state_file"
  done
}

disable_recording() {
  local free_bytes="$1"
  if [[ ! -f "$RECORDING_PAUSE_MARKER" ]]; then
    printf 'disabled_at=%s\nfree_bytes=%s\n' \
      "$(date --iso-8601=seconds)" "$free_bytes" > "$RECORDING_PAUSE_MARKER"
    alert CRITICAL "local video recording disabled at free=$(human_bytes "$free_bytes"); videos/videos_raw/annotations stop growing; detection, RTSP/live stream and uploader remain running; automatic recording resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
  else
    log WARNING "local video recording remains disabled: free=$(human_bytes "$free_bytes"), resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
  fi
  start_fast_resume_checks
}

enable_recording_if_ready() {
  local free_bytes="$1"
  [[ -f "$RECORDING_PAUSE_MARKER" ]] || {
    stop_fast_resume_checks
    return 0
  }
  if [[ "$free_bytes" -ge "$RESUME_TARGET_BYTES" ]]; then
    rm -f "$RECORDING_PAUSE_MARKER"
    stop_fast_resume_checks
    alert RECOVERED "free space reached $(human_bytes "$free_bytes"); local video recording enabled automatically; detection and live stream were never stopped"
  else
    log WARNING "local video recording remains disabled: free=$(human_bytes "$free_bytes"), resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
  fi
}

load_processed_files() {
  local destination="$1"
  if ! docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null | grep -qx true; then
    return 1
  fi
  docker exec -i "$CONTAINER" /opt/venv/bin/python - 2>/dev/null <<'PY' |
import json
import re
import socket

processed_path = "/smart-cam/kafka_sender/processed_files.log"
config_path = "/smart-cam/config/config.json"
receipts_path = "/smart-cam/data/upload_completed_types.json"
pattern = re.compile(r"^video_[0-9]{8}_[0-9]{6}_[0-9]{3}\.xml$")

with open(config_path, encoding="utf-8") as stream:
    config = json.load(stream)
upload = config.get("upload", {})
configured_device_id = str(config.get("device", {}).get("id", "")).strip()
host_device_id = socket.gethostname().strip()
device_id = configured_device_id or host_device_id
server = upload.get("server", {}) if isinstance(upload.get("server", {}), dict) else {}
root = str(server.get("root_dir", "/storage2/file/remote_sync")).rstrip("/")
namespace = str(upload.get("storage_namespace", "smart_cam")).strip("/")
remote_base = f"{root}/{namespace}/{device_id}" if device_id else ""
required = {
    name for name, default in (
        ("videos", False), ("videos_raw", True), ("annotations", True)
    ) if bool(upload.get(name, default))
}

try:
    with open(receipts_path, encoding="utf-8") as stream:
        receipts = json.load(stream)
except (FileNotFoundError, json.JSONDecodeError, OSError):
    receipts = {}

with open(processed_path, encoding="utf-8") as stream:
    for line in stream:
        name = line.strip()
        if not pattern.fullmatch(name):
            continue
        receipt = receipts.get(name, {})
        confirmed = set(receipt.get("types", []))
        if (
            device_id
            and (not configured_device_id or configured_device_id == host_device_id)
            and receipt.get("device_id") == device_id
            and receipt.get("remote_base") == remote_base
            and required.issubset(confirmed)
        ):
            print(name)
PY
    sort -u > "$destination"
  [[ -s "$destination" ]]
}

load_processed_scans() {
  local destination="$1"
  if ! docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null | grep -qx true; then
    return 1
  fi
  docker exec -i "$CONTAINER" /opt/venv/bin/python - 2>/dev/null <<'PY' |
import json
import os
import re
import socket

config_path = "/smart-cam/config/config.json"
processed_path = "/smart-cam/data/upload_processed_scans.log"
receipts_path = "/smart-cam/data/upload_completed_scans.json"
scan_root = "/smart-cam/data/scan"
pattern = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}_[0-9]{2}-[0-9]{2}-[0-9]{2}$")

with open(config_path, encoding="utf-8") as stream:
    config = json.load(stream)
upload = config.get("upload", {})
configured_device_id = str(config.get("device", {}).get("id", "")).strip()
host_device_id = socket.gethostname().strip()
if configured_device_id and configured_device_id != host_device_id:
    raise SystemExit(0)
device_id = configured_device_id or host_device_id
server = upload.get("server", {}) if isinstance(upload.get("server", {}), dict) else {}
root = str(server.get("root_dir", "/storage2/file/remote_sync")).rstrip("/")
namespace = str(upload.get("storage_namespace", "smart_cam")).strip("/")
remote_base = f"{root}/{namespace}/{device_id}" if device_id else ""
kafka_topic = str(upload.get("kafka_topic", "")).strip()

try:
    with open(receipts_path, encoding="utf-8") as stream:
        receipts = json.load(stream)
except (FileNotFoundError, json.JSONDecodeError, OSError):
    receipts = {}

try:
    with open(processed_path, encoding="utf-8") as stream:
        processed = {line.strip() for line in stream if pattern.fullmatch(line.strip())}
except OSError:
    processed = set()

for scan_id in sorted(processed):
    receipt = receipts.get(scan_id, {})
    image_count = int(receipt.get("image_count", 0))
    plan_included = receipt.get("patrol_plan_included") is True
    expected_files = image_count + 1 + (1 if plan_included else 0)
    valid = (
        device_id
        and receipt.get("device_id") == device_id
        and receipt.get("remote_base") == remote_base
        and receipt.get("remote_scan_path") == f"{remote_base}/scan/{scan_id}"
        and receipt.get("kafka_topic") == kafka_topic
        and receipt.get("sftp_confirmed") is True
        and receipt.get("kafka_confirmed") is True
        and image_count > 0
        and int(receipt.get("file_count", 0)) == expected_files
    )
    if not valid:
        continue
    try:
        with open(os.path.join(scan_root, scan_id, "meta.json"), encoding="utf-8") as stream:
            meta = json.load(stream)
        state = meta.get("state", {})
        valid = (
            meta.get("scan_id") == scan_id
            and state.get("state") == "completed"
            and int(state.get("total", 0)) == image_count
            and int(state.get("completed", -1)) == image_count
            and len(meta.get("images", [])) == image_count
            and bool(meta.get("patrol_plan")) == plan_included
        )
    except (OSError, ValueError, TypeError, json.JSONDecodeError):
        valid = False
    if valid:
        print(scan_id)
PY
    sort -u > "$destination"
  [[ -s "$destination" ]]
}

build_safe_bundles() {
  local processed="$1" output="$2" xml base raw annotated newest_ts path complete
  : > "$output"
  while IFS= read -r xml; do
    base="${xml%.xml}"
    raw="$DATA_ROOT/videos_raw/${base}_raw.mp4"
    annotated="$DATA_ROOT/videos/${base}_annotated.mp4"
    complete=1
    for path in "$raw" "$annotated" "$DATA_ROOT/annotations/${base}.xml"; do
      [[ -f "$path" ]] || {
        complete=0
        break
      }
      find "$path" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
        grep -q . || {
        complete=0
        break
      }
      if command -v fuser >/dev/null 2>&1 && fuser -s "$path" 2>/dev/null; then
        complete=0
        break
      fi
    done
    [[ "$complete" -eq 1 ]] || continue
    newest_ts="$(stat -c %Y "$raw" 2>/dev/null || printf 0)"
    printf '%s %s\n' "$newest_ts" "$base" >> "$output"
  done < "$processed"
  sort -n -u "$output" -o "$output"
}

delete_safe_bundles() {
  local bundles="$1" deleted_count=0 deleted_bytes=0 record base path size complete
  while IFS= read -r record; do
    base="${record#* }"
    complete=1
    for path in \
      "$DATA_ROOT/videos_raw/${base}_raw.mp4" \
      "$DATA_ROOT/videos/${base}_annotated.mp4" \
      "$DATA_ROOT/annotations/${base}.xml"; do
      [[ -f "$path" ]] || {
        complete=0
        break
      }
    done
    if [[ "$complete" -ne 1 ]]; then
      log WARNING "confirmed bundle became incomplete before deletion; skipped as a unit: ${base}"
      continue
    fi
    for path in \
      "$DATA_ROOT/videos_raw/${base}_raw.mp4" \
      "$DATA_ROOT/videos/${base}_annotated.mp4" \
      "$DATA_ROOT/annotations/${base}.xml"; do
      [[ -f "$path" ]] || continue
      size="$(stat -c %s "$path" 2>/dev/null || printf 0)"
      if rm -f -- "$path"; then
        deleted_count=$((deleted_count + 1))
        deleted_bytes=$((deleted_bytes + size))
        log INFO "deleted confirmed file (${size} bytes): ${path}"
      else
        log ERROR "failed to delete confirmed file: ${path}"
      fi
    done
  done < "$bundles"
  log INFO "safe cleanup result: deleted=${deleted_count} files bytes=${deleted_bytes} free=$(human_bytes "$(available_bytes)")"
}

delete_confirmed_scans() {
  local scan_id scan_dir resolved size confirmed
  local deleted_scans=0 deleted_bytes=0 unsafe

  confirmed="$(mktemp /tmp/confirmed-smartcam-scans.XXXXXX)"
  if ! load_processed_scans "$confirmed"; then
    rm -f -- "$confirmed"
    log INFO "no current-device strong scan upload receipts; scan cleanup skipped"
    return 0
  fi

  while IFS= read -r scan_id; do
    [[ "$scan_id" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}_[0-9]{2}-[0-9]{2}-[0-9]{2}$ ]] ||
      continue
    scan_dir="$DATA_ROOT/scan/$scan_id"
    [[ -d "$scan_dir" ]] || continue

    resolved="$(readlink -f -- "$scan_dir" 2>/dev/null || true)"
    [[ "$resolved" == "$DATA_ROOT/scan/$scan_id" ]] || {
      log WARNING "scan path failed safety check; skipped: ${scan_dir}"
      continue
    }

    # A processed marker is only written after every image/meta SFTP upload
    # and the Kafka delivery callback have succeeded. Re-check that this is a
    # completed scan and that it has aged out before deleting the directory.
    if ! python3 - "$scan_dir/meta.json" <<'PY'
import json
import sys
try:
    with open(sys.argv[1], encoding="utf-8") as stream:
        meta = json.load(stream)
    state = meta.get("state", {})
    images = meta.get("images", [])
    total = int(state.get("total", 0))
    valid = (
        state.get("state") == "completed"
        and total > 0
        and int(state.get("completed", -1)) == total
        and len(images) == total
    )
except (OSError, ValueError, TypeError, json.JSONDecodeError):
    valid = False
raise SystemExit(0 if valid else 1)
PY
    then
      log WARNING "confirmed scan has invalid/incomplete meta; skipped: ${scan_id}"
      continue
    fi

    if find "$scan_dir" -xdev -type f \
      ! -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
      grep -q .; then
      continue
    fi

    unsafe=0
    if command -v fuser >/dev/null 2>&1; then
      while IFS= read -r -d '' path; do
        if fuser -s "$path" 2>/dev/null; then
          log WARNING "confirmed scan file is still open; skipped scan=${scan_id} file=${path}"
          unsafe=1
          break
        fi
      done < <(find "$scan_dir" -xdev -type f -print0 2>/dev/null)
    fi
    [[ "$unsafe" -eq 0 ]] || continue

    size="$(du -sb "$scan_dir" 2>/dev/null | awk '{print $1}')"
    size="${size:-0}"
    if find "$scan_dir" -xdev -depth -delete 2>/dev/null &&
      [[ ! -e "$scan_dir" ]]; then
      deleted_scans=$((deleted_scans + 1))
      deleted_bytes=$((deleted_bytes + size))
      log INFO "deleted confirmed uploaded scan (${size} bytes): ${scan_dir}"
    else
      log ERROR "failed to delete confirmed uploaded scan: ${scan_dir}"
    fi
  done < "$confirmed"
  rm -f -- "$confirmed"

  log INFO "scan cleanup result: deleted=${deleted_scans} scans bytes=${deleted_bytes}"
}

delete_stale_incomplete_bundles() {
  local bases base filename raw annotated xml path
  local present safe size deleted_bundles=0 deleted_files=0 deleted_bytes=0
  bases="$(mktemp /tmp/userdata-incomplete-bases.XXXXXX)"

  find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f -name 'video_*_raw.mp4' \
    -printf '%f\n' 2>/dev/null |
    sed 's/_raw\.mp4$//' >> "$bases"
  find "$DATA_ROOT/videos" -maxdepth 1 -type f -name 'video_*_annotated.mp4' \
    -printf '%f\n' 2>/dev/null |
    sed 's/_annotated\.mp4$//' >> "$bases"
  find "$DATA_ROOT/annotations" -maxdepth 1 -type f -name 'video_*.xml' \
    -printf '%f\n' 2>/dev/null |
    sed 's/\.xml$//' >> "$bases"
  sort -u "$bases" -o "$bases"

  while IFS= read -r base; do
    [[ "$base" =~ ^video_[0-9]{8}_[0-9]{6}_[0-9]{3}$ ]] || continue
    raw="$DATA_ROOT/videos_raw/${base}_raw.mp4"
    annotated="$DATA_ROOT/videos/${base}_annotated.mp4"
    xml="$DATA_ROOT/annotations/${base}.xml"
    present=0
    [[ -f "$raw" ]] && present=$((present + 1))
    [[ -f "$annotated" ]] && present=$((present + 1))
    [[ -f "$xml" ]] && present=$((present + 1))
    [[ "$present" -gt 0 && "$present" -lt 3 ]] || continue

    # Recording creates members at slightly different times. Only reconcile
    # bundles when every existing member is old and no process has it open.
    safe=1
    for path in "$raw" "$annotated" "$xml"; do
      [[ -f "$path" ]] || continue
      find "$path" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
        grep -q . || {
        safe=0
        break
      }
      if command -v fuser >/dev/null 2>&1 && fuser -s "$path" 2>/dev/null; then
        log WARNING "incomplete bundle member is still open; skipped bundle=${base} file=${path}"
        safe=0
        break
      fi
    done
    [[ "$safe" -eq 1 ]] || continue

    for path in "$raw" "$annotated" "$xml"; do
      [[ -f "$path" ]] || continue
      size="$(stat -c %s "$path" 2>/dev/null || printf 0)"
      if rm -f -- "$path"; then
        deleted_files=$((deleted_files + 1))
        deleted_bytes=$((deleted_bytes + size))
        log INFO "deleted stale incomplete bundle member: ${path}"
      else
        log ERROR "failed to delete stale incomplete bundle member: ${path}"
      fi
    done
    deleted_bundles=$((deleted_bundles + 1))
  done < "$bases"
  rm -f -- "$bases"

  if [[ "$deleted_bundles" -gt 0 ]]; then
    log INFO "incomplete bundle reconciliation result: bundles=${deleted_bundles} files=${deleted_files} bytes=${deleted_bytes}"
  fi
}

delete_stale_zero_bundles() {
  local raw base path deleted_bundles=0 deleted_files=0
  while IFS= read -r -d '' raw; do
    # A newly opened recording may temporarily be zero bytes. Never touch a
    # recent or open file; only remove a stale, unusable bundle as one unit.
    find "$raw" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null | grep -q . || continue
    if command -v fuser >/dev/null 2>&1 && fuser -s "$raw" 2>/dev/null; then
      log WARNING "stale zero-byte raw is still open; skipped: ${raw}"
      continue
    fi

    base="$(basename "$raw" _raw.mp4)"
    for path in \
      "$DATA_ROOT/videos_raw/${base}_raw.mp4" \
      "$DATA_ROOT/videos/${base}_annotated.mp4" \
      "$DATA_ROOT/annotations/${base}.xml"; do
      [[ -f "$path" ]] || continue
      if rm -f -- "$path"; then
        deleted_files=$((deleted_files + 1))
        log INFO "deleted stale zero-byte bundle member: ${path}"
      else
        log ERROR "failed to delete stale zero-byte bundle member: ${path}"
      fi
    done
    deleted_bundles=$((deleted_bundles + 1))
  done < <(find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f \
    -name '*_raw.mp4' -size 0 -print0 2>/dev/null)

  if [[ "$deleted_bundles" -gt 0 ]]; then
    log INFO "stale zero-byte cleanup result: bundles=${deleted_bundles} files=${deleted_files}"
  fi
}

delete_stale_zero_part_bundles() {
  local part filename base final_raw sibling path size
  local safe part_count deleted_bundles=0 deleted_files=0 deleted_bytes=0
  declare -A checked_bases=()

  while IFS= read -r -d '' part; do
    [[ -f "$part" ]] || continue
    filename="$(basename "$part")"
    base="${filename%%_raw.mp4.part*}"
    [[ -n "$base" && -z "${checked_bases[$base]:-}" ]] || continue
    checked_bases["$base"]=1
    final_raw="$DATA_ROOT/videos_raw/${base}_raw.mp4"

    # If a valid final raw exists, these are harmless recorder leftovers.
    # Delete only old, zero-byte, unopened part files; preserve the bundle.
    if [[ -f "$final_raw" ]]; then
      while IFS= read -r -d '' sibling; do
        find "$sibling" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
          grep -q . || continue
        if command -v fuser >/dev/null 2>&1 && fuser -s "$sibling" 2>/dev/null; then
          log WARNING "stale zero-byte part is still open; skipped: ${sibling}"
          continue
        fi
        if rm -f -- "$sibling"; then
          deleted_files=$((deleted_files + 1))
          log INFO "deleted stale zero-byte recorder part: ${sibling}"
        fi
      done < <(find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f \
        -name "${base}_raw.mp4.part*.mp4" -size 0 -print0 2>/dev/null)
      continue
    fi

    # Without a final raw, only remove the incomplete bundle when every part
    # is old, zero bytes, and not held open by any process.
    safe=1
    part_count=0
    while IFS= read -r -d '' sibling; do
      part_count=$((part_count + 1))
      [[ "$(stat -c %s "$sibling" 2>/dev/null || printf 1)" -eq 0 ]] || {
        safe=0
        break
      }
      find "$sibling" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
        grep -q . || {
        safe=0
        break
      }
      if command -v fuser >/dev/null 2>&1 && fuser -s "$sibling" 2>/dev/null; then
        log WARNING "incomplete part bundle is still open; skipped: ${sibling}"
        safe=0
        break
      fi
    done < <(find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f \
      -name "${base}_raw.mp4.part*.mp4" -print0 2>/dev/null)

    [[ "$safe" -eq 1 && "$part_count" -gt 0 ]] || continue

    while IFS= read -r -d '' sibling; do
      size="$(stat -c %s "$sibling" 2>/dev/null || printf 0)"
      if rm -f -- "$sibling"; then
        deleted_files=$((deleted_files + 1))
        deleted_bytes=$((deleted_bytes + size))
        log INFO "deleted stale incomplete bundle part: ${sibling}"
      fi
    done < <(find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f \
      -name "${base}_raw.mp4.part*.mp4" -print0 2>/dev/null)

    for path in \
      "$DATA_ROOT/videos/${base}_annotated.mp4" \
      "$DATA_ROOT/annotations/${base}.xml"; do
      [[ -f "$path" ]] || continue
      size="$(stat -c %s "$path" 2>/dev/null || printf 0)"
      if rm -f -- "$path"; then
        deleted_files=$((deleted_files + 1))
        deleted_bytes=$((deleted_bytes + size))
        log INFO "deleted stale incomplete bundle member: ${path}"
      fi
    done
    deleted_bundles=$((deleted_bundles + 1))
  done < <(find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f \
    -name '*_raw.mp4.part*.mp4' -size 0 -print0 2>/dev/null)

  if [[ "$deleted_bundles" -gt 0 || "$deleted_files" -gt 0 ]]; then
    log INFO "stale part cleanup result: bundles=${deleted_bundles} files=${deleted_files} bytes=${deleted_bytes}"
  fi
}

main() {
  local free_bytes processed bundles
  recover_legacy_capture_pause
  free_bytes="$(available_bytes)"

  # Protect the disk before cleanup work. Only recording is disabled; continuous
  # detection, inference, RTSP/live output and upload remain available.
  if [[ "$free_bytes" -lt "$CRITICAL_BELOW_BYTES" ]]; then
    alert CRITICAL "free space below critical threshold: $(human_bytes "$free_bytes")"
  fi
  if [[ "$free_bytes" -lt "$PAUSE_BELOW_BYTES" ]]; then
    disable_recording "$free_bytes"
  fi
  enable_recording_if_ready "$free_bytes"

  processed="$(mktemp /tmp/userdata-processed.XXXXXX)"
  bundles="$(mktemp /tmp/userdata-bundles.XXXXXX)"
  trap 'rm -f "${processed:-}" "${bundles:-}"' EXIT

  if load_processed_files "$processed"; then
    build_safe_bundles "$processed" "$bundles"
    if [[ -s "$bundles" ]]; then
      delete_safe_bundles "$bundles"
    else
      log INFO "no confirmed old bundles are ready for safe cleanup"
    fi
  else
    alert ERROR "cannot read confirmed-upload list; fail-safe mode prevents all deletion"
  fi

  delete_confirmed_scans

  free_bytes="$(available_bytes)"
  if [[ "$free_bytes" -lt "$CRITICAL_BELOW_BYTES" ]]; then
    alert CRITICAL "free space below critical threshold: $(human_bytes "$free_bytes")"
  fi
  if [[ "$free_bytes" -lt "$PAUSE_BELOW_BYTES" ]]; then
    disable_recording "$free_bytes"
  fi
  enable_recording_if_ready "$(available_bytes)"
}

if [[ "${CLEANUP_SMARTCAM_LIB_ONLY:-0}" != "1" ]]; then
  main "$@"
fi
