mcp-nextcloud/nextcloud_mcp_server/migrations.py

"""Database migration utilities for nextcloud-mcp-server.

This module provides helper functions for managing Alembic database migrations
programmatically. It enables automatic migration on application startup and
provides CLI integration.
"""

import logging
import sqlite3
from pathlib import Path

from alembic.config import Config

import nextcloud_mcp_server.alembic as alembic_package
from alembic import command
from nextcloud_mcp_server.config import get_token_db_path

logger = logging.getLogger(__name__)


def get_alembic_config(database_path: str | Path | None = None) -> Config:
    """
    Get Alembic configuration for programmatic use.

    Works in both development and installed (Docker) modes by using
    package location instead of alembic.ini file.

    Args:
        database_path: Path to SQLite database file. If None, resolves via
                      config.get_token_db_path() (ephemeral tempfile unless
                      TOKEN_STORAGE_DB is set).

    Returns:
        Alembic Config object configured for the specified database
    """
    # Use package location (works in both editable and installed modes)
    if alembic_package.__file__ is None:
        raise RuntimeError("alembic package __file__ is None")
    script_location = Path(alembic_package.__file__).parent

    # Create config programmatically (no alembic.ini needed at runtime)
    config = Config()
    config.set_main_option("script_location", str(script_location))
    config.set_main_option("path_separator", "os")  # Suppress deprecation warning

    # Set database URL
    if database_path:
        db_path = Path(database_path).resolve()
    else:
        db_path = Path(get_token_db_path()).resolve()

    url = f"sqlite+aiosqlite:///{db_path}"
    config.set_main_option("sqlalchemy.url", url)

    logger.debug("Alembic script location: %s", script_location)
    logger.debug("Database: %s", db_path)

    return config


def upgrade_database(
    database_path: str | Path | None = None, revision: str = "head"
) -> None:
    """
    Upgrade database to a specific revision.

    Args:
        database_path: Path to SQLite database file
        revision: Target revision (default: "head" for latest)
    """
    config = get_alembic_config(database_path)
    logger.info("Upgrading database to revision: %s", revision)
    command.upgrade(config, revision)
    logger.info("Database upgrade completed successfully")


def downgrade_database(
    database_path: str | Path | None = None, revision: str = "-1"
) -> None:
    """
    Downgrade database to a specific revision.

    Args:
        database_path: Path to SQLite database file
        revision: Target revision (default: "-1" for previous version)
    """
    config = get_alembic_config(database_path)
    logger.warning("Downgrading database to revision: %s", revision)
    command.downgrade(config, revision)
    logger.info("Database downgrade completed successfully")


def get_current_revision(database_path: str | Path | None = None) -> str | None:
    """
    Get the current database revision by directly querying the alembic_version table.

    Args:
        database_path: Path to SQLite database file

    Returns:
        Current revision ID or None if not versioned
    """

    if database_path is None:
        database_path = get_token_db_path()

    db_path = Path(database_path).resolve()

    if not db_path.exists():
        logger.debug("Database does not exist: %s", db_path)
        return None

    try:
        # Query alembic_version table directly
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()

        # Check if alembic_version table exists
        cursor.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='alembic_version'"
        )
        has_table = cursor.fetchone() is not None

        if not has_table:
            conn.close()
            return None

        # Get current version
        cursor.execute("SELECT version_num FROM alembic_version")
        row = cursor.fetchone()
        conn.close()

        return row[0] if row else None

    except Exception as e:
        logger.error("Failed to get current revision: %s", e)
        return None


def stamp_database(
    database_path: str | Path | None = None, revision: str = "head"
) -> None:
    """
    Stamp database with a specific revision without running migrations.

    This is useful for marking existing databases that were created before
    Alembic was introduced. It tells Alembic "this database is at revision X"
    without actually running the migration.

    Args:
        database_path: Path to SQLite database file
        revision: Revision to stamp (default: "head" for latest)
    """
    config = get_alembic_config(database_path)
    logger.info("Stamping database with revision: %s", revision)
    command.stamp(config, revision)
    logger.info("Database stamped successfully")


def show_migration_history(database_path: str | Path | None = None) -> None:
    """
    Display migration history.

    Args:
        database_path: Path to SQLite database file
    """
    config = get_alembic_config(database_path)
    command.history(config, verbose=True)


def create_migration(message: str, autogenerate: bool = False) -> None:
    """
    Create a new migration script.

    Args:
        message: Description of the migration
        autogenerate: Whether to attempt auto-generation (requires SQLAlchemy models)

    Note:
        Since we don't use SQLAlchemy models, autogenerate will be disabled
        and migrations must be written manually.
    """
    config = get_alembic_config()
    logger.info("Creating new migration: %s", message)

    if autogenerate:
        logger.warning(
            "Auto-generation is not supported (no SQLAlchemy models). "
            "Migration will be created with empty upgrade/downgrade functions."
        )

    command.revision(config, message=message, autogenerate=False)
    logger.info("Migration created successfully. Edit the file to add SQL statements.")