ObsiGate/backend/services/backups.py

"""Backup inspection services shared by REST routes and the AI tool layer.

Single source of truth for locating, listing and diffing the timestamped
backups created before each file mutation. The write-side (creating a backup)
stays in the route layer; only the read/inspection logic lives here so both
``/api/file/{vault}/backups``, ``/api/file/{vault}/diff`` and the tools
``list_backups`` / ``diff_backup`` behave identically.
"""

from __future__ import annotations

import difflib
import logging
import os
import shutil
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from backend.services.errors import ServiceError
from backend.services.paths import resolve_safe_path
from backend.services.vaults import get_vault_root

logger = logging.getLogger("obsigate.services.backups")

# Default number of backups kept per file (matches ``max_backups_per_file``).
DEFAULT_MAX_BACKUPS = 10


def _default_max_backups() -> int:
    """Read ``max_backups_per_file`` from app config (lazy, best-effort)."""
    try:
        from backend.routers.config import _load_config  # ROADMAP #85 T7 — déménagé depuis backend.main

        return int(_load_config().get("max_backups_per_file", DEFAULT_MAX_BACKUPS))
    except Exception:  # pragma: no cover - config unavailable
        return DEFAULT_MAX_BACKUPS


def get_backup_dir(vault_name: str, relative_path: str) -> Path:
    """Return the directory where backups for a specific file are stored.

    Resolves relative backup paths against the SPECIFIC vault's directory,
    matching the backup-creation logic exactly.
    """
    from backend.indexer import get_vault_data

    backup_root = Path(os.environ.get("OBSIGATE_BACKUP_DIR", ".obsigate-backup"))
    if not backup_root.is_absolute():
        vault_data = get_vault_data(vault_name)
        if vault_data:
            backup_root = Path(vault_data["path"]) / backup_root
    return backup_root / vault_name / Path(relative_path).parent


def create_backup(
    file_path: Path,
    vault_name: str,
    relative_path: str,
    *,
    max_backups: int | None = None,
) -> Path | None:
    """Create a timestamped backup of a file before it is modified.

    Backups are stored as ``{backup_root}/{vault}/{relative_path}.{ts}.bak``.
    Missing or unreadable files are skipped silently (returns ``None``) so a
    backup failure never blocks the caller's mutation.

    Args:
        file_path: Absolute path of the file to back up.
        vault_name: Name of the vault the file belongs to.
        relative_path: Vault-relative path (used to mirror the tree).
        max_backups: Number of backups to keep per file. ``None`` reads
            ``max_backups_per_file`` from the app config.

    Returns:
        The created backup path, or ``None`` when skipped.
    """
    try:
        if not file_path.exists() or not file_path.is_file():
            logger.debug(f"Backup skipped: file not found {file_path}")
            return None

        backup_dir = get_backup_dir(vault_name, relative_path)
        backup_dir.mkdir(parents=True, exist_ok=True)
        backup_path = backup_dir / f"{file_path.name}.{int(time.time())}.bak"
        shutil.copy2(file_path, backup_path)
        logger.info(f"Backup created: {relative_path} -> {backup_path}")

        keep = max_backups if max_backups is not None else _default_max_backups()
        all_backups = sorted(
            [
                f
                for f in backup_dir.iterdir()
                if f.is_file() and f.name.startswith(file_path.name + ".") and f.name.endswith(".bak")
            ],
            key=lambda f: f.stat().st_mtime,
            reverse=True,
        )
        for old in all_backups[keep:]:
            old.unlink()
            logger.debug(f"Auto-cleanup: removed old backup {old.name}")
        return backup_path
    except Exception as e:
        logger.warning(f"Failed to backup {relative_path} (vault={vault_name}): {e}", exc_info=True)
        return None


def list_backup_files(vault_name: str, relative_path: str) -> list[dict[str, Any]]:
    """List all backup files for a vault file, sorted newest first.

    Backup filename format: ``{original_filename}.{timestamp}.bak``.

    Returns a list of dicts with ``timestamp``, ``datetime``, ``size`` and
    ``filename``. Missing or unreadable backup directories yield an empty list.
    """
    backup_dir = get_backup_dir(vault_name, relative_path)
    if not backup_dir.exists():
        return []

    original_name = Path(relative_path).name
    prefix = original_name + "."
    backups: list[dict[str, Any]] = []

    try:
        dir_entries = list(backup_dir.iterdir())
    except PermissionError:
        logger.warning(f"Permission denied reading backup dir: {backup_dir}")
        return []
    except OSError as e:
        logger.error(f"Error reading backup dir {backup_dir}: {e}")
        return []

    for f in dir_entries:
        try:
            if not f.is_file():
                continue
            name = f.name
            if not name.startswith(prefix) or not name.endswith(".bak"):
                continue
            ts_part = name[len(prefix):-len(".bak")]
            try:
                ts = int(ts_part)
            except (ValueError, TypeError):
                continue
            try:
                dt = datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
            except (OSError, OverflowError, ValueError) as ts_err:
                logger.warning(f"Skipping backup with invalid timestamp {ts}: {ts_err}")
                continue
            st = f.stat()
            backups.append({
                "timestamp": ts,
                "datetime": dt,
                "size": st.st_size,
                "filename": name,
            })
        except Exception as entry_err:
            logger.warning(f"Skipping unreadable backup entry {f}: {entry_err}")
            continue

    backups.sort(key=lambda b: b["timestamp"], reverse=True)
    return backups


def diff_backup(
    vault_name: str,
    path: str,
    version: int,
    compare_with: int | None = None,
) -> dict[str, Any]:
    """Generate a unified diff between a backup version and another version or the current file.

    Args:
        vault_name: Name of the vault.
        path: Relative path of the file within the vault.
        version: Timestamp of the backup used as the old/left side.
        compare_with: Optional timestamp of another backup as the new/right
            side. If omitted, the current file on disk is used.

    Returns:
        Dict with ``vault``, ``path``, ``version``, ``compare_with``, ``diff``,
        ``left_content`` and ``right_content``.

    Raises:
        ServiceError: ``not_found`` (404) when the file or a backup is missing,
            ``read_error`` (500) when a file cannot be read.
    """
    root = get_vault_root(vault_name)
    file_path = resolve_safe_path(root, path)

    if not file_path.exists() or not file_path.is_file():
        raise ServiceError(
            f"File not found: {path}",
            code="not_found",
            status=404,
            details={"vault": vault_name, "path": path},
        )

    original_name = Path(path).name

    def _read_backup(ts: int) -> tuple[str, str]:
        backup_dir = get_backup_dir(vault_name, path)
        backup_path = backup_dir / f"{original_name}.{ts}.bak"
        if not backup_path.exists():
            raise ServiceError(
                f"Backup version {ts} not found for {path}",
                code="not_found",
                status=404,
                details={"vault": vault_name, "path": path, "version": ts},
            )
        try:
            content = backup_path.read_text(encoding="utf-8", errors="replace")
        except Exception as e:
            raise ServiceError(
                f"Failed to read backup {ts}: {e}",
                code="read_error",
                status=500,
                details={"vault": vault_name, "path": path, "version": ts},
            ) from e
        dt = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
        return content, f"{path}@{dt}"

    try:
        left_content, left_label = _read_backup(version)

        if compare_with is not None:
            right_content, right_label = _read_backup(compare_with)
        else:
            try:
                right_content = file_path.read_text(encoding="utf-8", errors="replace")
            except Exception as e:
                raise ServiceError(
                    f"Failed to read current file: {e}",
                    code="read_error",
                    status=500,
                    details={"vault": vault_name, "path": path},
                ) from e
            right_label = f"{path} (current)"

        left_lines = left_content.splitlines(keepends=True)
        right_lines = right_content.splitlines(keepends=True)

        diff_lines = list(difflib.unified_diff(
            left_lines, right_lines,
            fromfile=left_label, tofile=right_label,
        ))

        return {
            "vault": vault_name,
            "path": path,
            "version": version,
            "compare_with": compare_with,
            "diff": "".join(diff_lines),
            "left_content": left_content,
            "right_content": right_content,
        }
    except ServiceError:
        raise
    except Exception as e:
        logger.error(f"Error generating diff for {vault_name}/{path}: {type(e).__name__}: {e}", exc_info=True)
        raise ServiceError(f"Erreur lors de la génération du diff: {e!s}", code="read_error", status=500) from e