AirlockTools/utils/versionchecker.py

#    Copyright (C) 2025 James Brotosky, Brandon Wickline
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as published
#    by the Free Software Foundation, either version 3 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""
Version checking and update notification system for Loxide.

Checks against Gitea releases at:
https://git.racooncity.org/brotoskyj/AirlockTools/releases
"""

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
import json
import logging
from pathlib import Path
import re
import threading
from typing import Callable, Optional

import requests

logger = logging.getLogger(__name__)

# Current application version - UPDATE THIS ON EACH RELEASE
__version__ = "1.0.0"

# Gitea release API configuration
GITEA_API_BASE = "https://git.racooncity.org/api/v1"
REPO_OWNER = "brotoskyj"
REPO_NAME = "AirlockTools"
RELEASES_URL = f"{GITEA_API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/releases"
RELEASES_PAGE_URL = f"https://git.racooncity.org/{REPO_OWNER}/{REPO_NAME}/releases"

# How often to check for updates (in hours)
CHECK_INTERVAL_HOURS = 24


@dataclass
class ReleaseInfo:
    """Information about a release."""

    tag_name: str
    version: tuple  # Parsed semantic version (major, minor, patch)
    name: str
    body: str  # Release notes
    published_at: datetime
    html_url: str
    download_url: Optional[str] = None  # URL to download the release asset
    is_prerelease: bool = False


@dataclass
class UpdateCheckResult:
    """Result of an update check."""

    current_version: str
    latest_version: Optional[str]
    update_available: bool
    release_info: Optional[ReleaseInfo]
    error: Optional[str] = None


def parse_version(version_str: str) -> Optional[tuple]:
    """
    Parse a version string into a comparable tuple.
    Supports formats: v1.2.3, 1.2.3, v1.2, 1.2

    Returns (major, minor, patch) tuple or None if parsing fails.
    """
    if not version_str:
        return None

    # Strip 'v' prefix if present
    clean = version_str.lstrip("vV").strip()

    # Match semantic version pattern
    match = re.match(r"^(\d+)(?:\.(\d+))?(?:\.(\d+))?", clean)
    if not match:
        return None

    major = int(match.group(1))
    minor = int(match.group(2)) if match.group(2) else 0
    patch = int(match.group(3)) if match.group(3) else 0

    return (major, minor, patch)


def compare_versions(v1: tuple, v2: tuple) -> int:
    """
    Compare two version tuples.

    Returns:
        -1 if v1 < v2
         0 if v1 == v2
         1 if v1 > v2
    """
    for a, b in zip(v1, v2):
        if a < b:
            return -1
        if a > b:
            return 1
    return 0


def get_current_version() -> str:
    """Get the current application version."""
    return __version__


def _parse_release_response(release_data: dict) -> Optional[ReleaseInfo]:
    """Parse a release from Gitea API response."""
    try:
        tag_name = release_data.get("tag_name", "")
        version = parse_version(tag_name)
        if not version:
            logger.debug(f"Could not parse version from tag: {tag_name}")
            return None

        # Parse published date
        published_str = release_data.get("published_at", "")
        try:
            published_at = datetime.fromisoformat(published_str.replace("Z", "+00:00"))
        except (ValueError, AttributeError):
            published_at = datetime.now(UTC)

        # Get download URL from assets if available
        download_url = None
        assets = release_data.get("assets", [])
        for asset in assets:
            # Prefer .exe or .zip files
            name = asset.get("name", "").lower()
            if name.endswith((".exe", ".zip", ".msi")):
                download_url = asset.get("browser_download_url")
                break

        return ReleaseInfo(
            tag_name=tag_name,
            version=version,
            name=release_data.get("name", tag_name),
            body=release_data.get("body", ""),
            published_at=published_at,
            html_url=release_data.get("html_url", RELEASES_PAGE_URL),
            download_url=download_url,
            is_prerelease=release_data.get("prerelease", False),
        )
    except Exception as e:
        logger.warning(f"Failed to parse release data: {e}")
        return None


def fetch_latest_release(
    include_prerelease: bool = False, timeout: int = 10
) -> Optional[ReleaseInfo]:
    """
    Fetch the latest release from Gitea.

    Args:
        include_prerelease: Whether to include pre-release versions
        timeout: Request timeout in seconds

    Returns:
        ReleaseInfo for the latest release, or None if fetch fails
    """
    try:
        response = requests.get(
            RELEASES_URL,
            params={"limit": 10},  # Get last 10 releases to find latest stable
            timeout=timeout,
            headers={"Accept": "application/json"},
        )
        response.raise_for_status()

        releases = response.json()
        if not releases:
            logger.debug("No releases found")
            return None

        # Find the latest release (first non-prerelease if we're excluding them)
        for release_data in releases:
            release_info = _parse_release_response(release_data)
            if release_info is None:
                continue

            if include_prerelease or not release_info.is_prerelease:
                return release_info

        # If all are prereleases and we're excluding them, return the first one anyway
        # but log a warning
        if releases:
            logger.debug("All releases are pre-releases")
            return _parse_release_response(releases[0])

        return None

    except requests.exceptions.Timeout:
        logger.warning("Timeout fetching releases from Gitea")
        return None
    except requests.exceptions.RequestException as e:
        logger.warning(f"Failed to fetch releases: {e}")
        return None
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning(f"Failed to parse release response: {e}")
        return None


def check_for_updates(include_prerelease: bool = False) -> UpdateCheckResult:
    """
    Check if a newer version is available.

    Args:
        include_prerelease: Whether to consider pre-release versions

    Returns:
        UpdateCheckResult with the check results
    """
    current = get_current_version()
    current_parsed = parse_version(current)

    if not current_parsed:
        return UpdateCheckResult(
            current_version=current,
            latest_version=None,
            update_available=False,
            release_info=None,
            error="Could not parse current version",
        )

    release_info = fetch_latest_release(include_prerelease=include_prerelease)

    if release_info is None:
        return UpdateCheckResult(
            current_version=current,
            latest_version=None,
            update_available=False,
            release_info=None,
            error="Could not fetch release information",
        )

    is_newer = compare_versions(release_info.version, current_parsed) > 0

    return UpdateCheckResult(
        current_version=current,
        latest_version=release_info.tag_name,
        update_available=is_newer,
        release_info=release_info,
    )


class VersionChecker:
    """
    Background version checker that periodically checks for updates
    and can notify the application when updates are available.
    """

    def __init__(
        self,
        cache_dir: Optional[Path] = None,
        check_interval_hours: int = CHECK_INTERVAL_HOURS,
        on_update_available: Optional[Callable[[UpdateCheckResult], None]] = None,
    ):
        """
        Initialize the version checker.

        Args:
            cache_dir: Directory to store last check timestamp
            check_interval_hours: Hours between automatic checks
            on_update_available: Callback when update is available
        """
        self.cache_dir = cache_dir
        self.check_interval = timedelta(hours=check_interval_hours)
        self.on_update_available = on_update_available
        self._last_check: Optional[datetime] = None
        self._last_result: Optional[UpdateCheckResult] = None
        self._check_thread: Optional[threading.Thread] = None
        self._dismissed_version: Optional[str] = None

        # Load cached state
        self._load_cache()

    @property
    def cache_file(self) -> Optional[Path]:
        if self.cache_dir:
            return self.cache_dir / "version_check_cache.json"
        return None

    def _load_cache(self) -> None:
        """Load cached check state."""
        if not self.cache_file or not self.cache_file.exists():
            return

        try:
            with open(self.cache_file, "r") as f:
                data = json.load(f)

            if "last_check" in data:
                self._last_check = datetime.fromisoformat(data["last_check"])
            # Don't load dismissed_version - dismiss is session-only

        except (json.JSONDecodeError, ValueError, OSError) as e:
            logger.debug(f"Could not load version check cache: {e}")

    def _save_cache(self) -> None:
        """Save check state to cache."""
        if not self.cache_file:
            return

        try:
            self.cache_file.parent.mkdir(parents=True, exist_ok=True)
            data = {}
            if self._last_check:
                data["last_check"] = self._last_check.isoformat()
            # Don't save dismissed_version - dismiss is session-only

            with open(self.cache_file, "w") as f:
                json.dump(data, f)

        except OSError as e:
            logger.debug(f"Could not save version check cache: {e}")

    def should_check(self) -> bool:
        """Determine if enough time has passed to check again."""
        if self._last_check is None:
            return True

        elapsed = datetime.now(UTC) - self._last_check
        return elapsed >= self.check_interval

    def check_now(
        self, force: bool = False, include_prerelease: bool = False
    ) -> UpdateCheckResult:
        """
        Check for updates immediately.

        Args:
            force: Check even if recently checked
            include_prerelease: Include pre-release versions

        Returns:
            UpdateCheckResult
        """
        if not force and not self.should_check() and self._last_result:
            return self._last_result

        result = check_for_updates(include_prerelease=include_prerelease)
        self._last_check = datetime.now(UTC)
        self._last_result = result
        self._save_cache()

        # Notify if update available and not dismissed
        if (
            result.update_available
            and self.on_update_available
            and result.latest_version != self._dismissed_version
        ):
            self.on_update_available(result)

        return result

    def check_async(
        self, force: bool = False, include_prerelease: bool = False
    ) -> None:
        """
        Check for updates in background thread.

        Args:
            force: Check even if recently checked
            include_prerelease: Include pre-release versions
        """
        if self._check_thread and self._check_thread.is_alive():
            return  # Already checking

        if not force and not self.should_check():
            return  # Too soon to check again

        def _check():
            try:
                self.check_now(force=True, include_prerelease=include_prerelease)
            except Exception as e:
                logger.debug(f"Background version check failed: {e}")

        self._check_thread = threading.Thread(target=_check, daemon=True)
        self._check_thread.start()

    def dismiss_update(self, version: str) -> None:
        """
        Dismiss update notification for a specific version.
        Only lasts for the current session - will nag again on next startup.

        Args:
            version: Version to dismiss (e.g., "v1.2.3")
        """
        # Session-only dismiss - don't save to cache
        self._dismissed_version = version

    def clear_dismissed(self) -> None:
        """Clear the dismissed version so user will be nagged again."""
        self._dismissed_version = None

    def get_last_result(self) -> Optional[UpdateCheckResult]:
        """Get the result of the last check."""
        return self._last_result


# Global instance for easy access
_global_checker: Optional[VersionChecker] = None


def get_version_checker(
    cache_dir: Optional[Path] = None,
    on_update_available: Optional[Callable[[UpdateCheckResult], None]] = None,
) -> VersionChecker:
    """
    Get or create the global version checker instance.

    Args:
        cache_dir: Directory for caching (only used on first call)
        on_update_available: Callback for updates (only used on first call)

    Returns:
        The global VersionChecker instance
    """
    global _global_checker

    if _global_checker is None:
        _global_checker = VersionChecker(
            cache_dir=cache_dir,
            on_update_available=on_update_available,
        )

    return _global_checker


def format_update_message(result: UpdateCheckResult, short: bool = False) -> str:
    """
    Format a human-readable update message.

    Args:
        result: The update check result
        short: Whether to use a short format

    Returns:
        Formatted message string
    """
    if not result.update_available:
        return f"✅ Loxide is up to date (v{result.current_version})"

    if short:
        return f"🆕 Update available: {result.latest_version}"

    msg = f"🆕 Loxide {result.latest_version} is available! (current: v{result.current_version})"

    if result.release_info:
        msg += f"\n📥 Download: {result.release_info.html_url}"

        # Include release notes preview if available
        if result.release_info.body:
            notes = result.release_info.body.strip()
            # Truncate if too long
            if len(notes) > 200:
                notes = notes[:200] + "..."
            msg += f"\n\n📋 Release Notes:\n{notes}"

    return msg


# ---------------------------------------------------------------------------
# Textual TUI Integration
# ---------------------------------------------------------------------------


def create_update_notifier(
    app, cache_dir: Optional[Path] = None, nag_on_startup: bool = True
):
    """
    Create a version checker that notifies via Textual toast notifications.

    This should be called after the Textual app is created.

    Args:
        app: The Textual App instance
        cache_dir: Directory for caching check state
        nag_on_startup: Always show notification on startup if update available

    Returns:
        The VersionChecker instance
    """

    def on_update_available(result: UpdateCheckResult):
        """Callback when update is available - show toast notification."""
        try:
            msg = f"🆕 Update available: {result.latest_version}\nGo to Settings to download"
            try:
                app.notify(
                    msg, title="Loxide Update Available", severity="warning", timeout=15
                )
            except RuntimeError:
                app.call_from_thread(
                    app.notify,
                    msg,
                    title="Loxide Update Available",
                    severity="warning",
                    timeout=15,
                )
        except Exception as e:
            logger.debug(f"Could not show update notification: {e}")

    checker = get_version_checker(
        cache_dir=cache_dir,
        on_update_available=on_update_available,
    )

    # Store checker on app so Loxide.on_mount can use it
    if nag_on_startup:
        app._version_checker = checker
        app._version_nag_shown = False

    return checker


def check_for_updates_startup(
    cache_dir: Optional[Path] = None,
) -> Optional[UpdateCheckResult]:
    """
    Check for updates during application startup.

    This performs a synchronous check but respects the cache interval,
    so it will only actually query the network once per CHECK_INTERVAL_HOURS.

    Returns the result if an update is available, None otherwise.

    Example usage:
        result = check_for_updates_startup(cache_dir)
        if result and result.update_available:
            print(format_update_message(result))
    """
    checker = get_version_checker(cache_dir=cache_dir)

    # Only check if enough time has passed (uses cache)
    if not checker.should_check():
        result = checker.get_last_result()
        if result and result.update_available:
            return result
        return None

    result = checker.check_now(force=False)
    if result.update_available:
        return result
    return None