diff --git a/db/init-repology.sql b/db/init-repology.sql index 2ae43fc..12cf15b 100644 --- a/db/init-repology.sql +++ b/db/init-repology.sql @@ -1,13 +1,118 @@ --- Create database and user -CREATE DATABASE repology; +-- ============================================================================ +-- Repology Database Initialization with Rotation Support +-- ============================================================================ + +-- Create repology user CREATE USER repology WITH PASSWORD 'repology'; + +-- Create control database for metadata and rotation management +CREATE DATABASE repology_control; + +-- Create two Repology databases for rotation +CREATE DATABASE repology; +CREATE DATABASE repology_replica; + +-- Grant permissions on all databases +GRANT ALL ON DATABASE repology_control TO repology; GRANT ALL ON DATABASE repology TO repology; +GRANT ALL ON DATABASE repology_replica TO repology; +-- ============================================================================ +-- Setup Main Database +-- ============================================================================ \connect repology -- Schema permissions GRANT CREATE ON SCHEMA public TO PUBLIC; -- Enable extensions -CREATE EXTENSION pg_trgm; -CREATE EXTENSION libversion; +CREATE EXTENSION IF NOT EXISTS pg_trgm; +CREATE EXTENSION IF NOT EXISTS libversion; + +-- ============================================================================ +-- Setup Replica Database +-- ============================================================================ +\connect repology_replica + +-- Schema permissions +GRANT CREATE ON SCHEMA public TO PUBLIC; + +-- Enable extensions +CREATE EXTENSION IF NOT EXISTS pg_trgm; +CREATE EXTENSION IF NOT EXISTS libversion; + +-- ============================================================================ +-- Setup Control Database +-- ============================================================================ +\connect repology_control + +-- Create metadata table to track active database +CREATE TABLE IF NOT EXISTS repology_control ( + id SERIAL PRIMARY KEY, + active_database VARCHAR(50) NOT NULL, + last_updated TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + last_rotation TIMESTAMP WITH TIME ZONE, + CONSTRAINT check_database_name CHECK (active_database IN ('repology', 'repology_replica')) +); + +-- Grant permissions on control table +GRANT SELECT, INSERT, UPDATE ON repology_control TO repology; +GRANT USAGE, SELECT ON SEQUENCE repology_control_id_seq TO repology; + +-- Initialize with main database as default active +INSERT INTO repology_control (active_database) +VALUES ('repology') +ON CONFLICT DO NOTHING; + +-- Create view for easy active database lookup +CREATE OR REPLACE VIEW repology_active_config AS +SELECT + active_database, + last_updated, + last_rotation +FROM repology_control +ORDER BY id DESC +LIMIT 1; + +-- Grant view access +GRANT SELECT ON repology_active_config TO repology; + +-- Create function to get active database name +CREATE OR REPLACE FUNCTION get_active_repology_database() +RETURNS VARCHAR(50) AS $$ +DECLARE + db_name VARCHAR(50); +BEGIN + SELECT active_database INTO db_name + FROM repology_active_config; + RETURN db_name; +END; +$$ LANGUAGE plpgsql SECURITY DEFINER; + +-- Grant execute permission +GRANT EXECUTE ON FUNCTION get_active_repology_database() TO repology; + +-- Create table to track rotation history +CREATE TABLE IF NOT EXISTS repology_rotation_history ( + id SERIAL PRIMARY KEY, + from_database VARCHAR(50), + to_database VARCHAR(50), + rotation_timestamp TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + success BOOLEAN DEFAULT true, + validation_status TEXT, + error_message TEXT, + duration_seconds INTEGER +); + +-- Grant permissions on history table +GRANT SELECT, INSERT ON repology_rotation_history TO repology; +GRANT USAGE, SELECT ON SEQUENCE repology_rotation_history_id_seq TO repology; + +-- ============================================================================ +-- Verification +-- ============================================================================ + +-- Display configuration +SELECT 'Database initialization complete!' AS status; +SELECT 'Active database: ' || active_database AS configuration +FROM repology_active_config; diff --git a/db/load-repology.sh b/db/load-repology.sh old mode 100644 new mode 100755 index 4b75fad..4b753c5 --- a/db/load-repology.sh +++ b/db/load-repology.sh @@ -1,12 +1,16 @@ #!/bin/bash set -euo pipefail +# ============================================================================ +# Repology Database Load Script with Rotation +# ============================================================================ + # Database configuration from environment variables with defaults DB_HOST="${POSTGRES_HOST:-localhost}" DB_PORT="${POSTGRES_PORT:-5432}" -DB_NAME="${POSTGRES_DB:-repology}" DB_USER="${POSTGRES_USER:-repology}" DB_PASSWORD="${POSTGRES_PASSWORD:-}" +CONTROL_DB="${CONTROL_DB:-repology_control}" # Proxy configuration (optional) PROXY_URL="${PROXY_URL:-}" @@ -14,8 +18,141 @@ PROXY_URL="${PROXY_URL:-}" # Dump URL configuration (optional) DUMP_URL="${DUMP_URL:-https://dumps.repology.org/repology-database-dump-latest.sql.zst}" -echo "Downloading Repology dump..." -echo "Source: $DUMP_URL" +# Start time for duration tracking +START_TIME=$(date +%s) + +# ============================================================================ +# Helper Functions +# ============================================================================ + +log() { + echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" +} + +error() { + echo "[$(date '+%Y-%m-%d %H:%M:%S')] ERROR: $*" >&2 +} + +# Execute SQL query and return result +execute_sql() { + local database=$1 + local query=$2 + + if [ -n "$DB_PASSWORD" ]; then + export PGPASSWORD="$DB_PASSWORD" + fi + + psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$database" -t -A -c "$query" 2>/dev/null || echo "" + + unset PGPASSWORD +} + +# Get active database from control table +get_active_database() { + local active_db=$(execute_sql "$CONTROL_DB" "SELECT active_database FROM repology_active_config;") + + if [ -z "$active_db" ]; then + error "Failed to query active database from control table" + exit 1 + fi + + echo "$active_db" +} + +# Get inactive database (the one to load data into) +get_inactive_database() { + local active_db=$1 + + if [ "$active_db" = "repology" ]; then + echo "repology_replica" + else + echo "repology" + fi +} + +# Switch active database in control table +switch_active_database() { + local from_db=$1 + local to_db=$2 + local validation_status=$3 + + log "Switching active database: $from_db -> $to_db" + + if [ -n "$DB_PASSWORD" ]; then + export PGPASSWORD="$DB_PASSWORD" + fi + + # Calculate duration + local end_time=$(date +%s) + local duration=$((end_time - START_TIME)) + + # Update control table + psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$CONTROL_DB" -v ON_ERROR_STOP=1 < pg_backend_pid(); + +-- Drop and recreate database +DROP DATABASE IF EXISTS $TARGET_DB; +CREATE DATABASE $TARGET_DB; +GRANT ALL ON DATABASE $TARGET_DB TO $DB_USER; +EOF + +# Connect to target database and enable extensions +log "Setting up extensions in $TARGET_DB..." +psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$TARGET_DB" < str: + """Get database DSN string for a specific database.""" + return f"postgresql://{self.dbuser}:{self.password}@{self.host}:{self.port}/{database}" + @property - def dsn(self) -> str: - """Get database DSN string.""" - return f"postgresql://{self.dbuser}:{self.password}@{self.host}:{self.port}/{self.database}" + def control_dsn(self) -> str: + """Get control database DSN string.""" + return self.get_dsn(self.control_database) class APIConfig(BaseSettings): diff --git a/src/repologyapi/database/connection.py b/src/repologyapi/database/connection.py index 5ce918b..906cafc 100644 --- a/src/repologyapi/database/connection.py +++ b/src/repologyapi/database/connection.py @@ -12,40 +12,183 @@ class DatabaseConnection: - """Database connection manager with connection pooling.""" + """Database connection manager with connection pooling and blue-green rotation support.""" def __init__(self, config: DatabaseConfig): self.config = config self._pool: Optional[asyncpg.Pool] = None + self._current_database: Optional[str] = None + self._refresh_task: Optional[asyncio.Task] = None + self._shutdown_event: asyncio.Event = asyncio.Event() + + @property + def database(self) -> str: + """Get currently active database name.""" + return self._current_database or "" + + async def query_active_database(self) -> str: + """Query control database to get currently active database name.""" + try: + # Create temporary connection to control database + conn = await asyncpg.connect(dsn=self.config.control_dsn) + try: + active_db = await conn.fetchval( + "SELECT active_database FROM repology_active_config" + ) + if not active_db: + raise RuntimeError("No active database configured in control table") + return active_db + finally: + await conn.close() + except Exception as e: + logger.error(f"Failed to query active database: {e}") + raise + + async def _create_pool(self, database: str) -> asyncpg.Pool: + """Create a connection pool for the specified database.""" + dsn = self.config.get_dsn(database) + return await asyncpg.create_pool( + dsn=dsn, + min_size=self.config.min_size, + max_size=self.config.max_size, + command_timeout=self.config.command_timeout, + server_settings={ + 'application_name': 'repologyapi', + 'search_path': 'public', + } + ) async def connect(self) -> None: - """Create database connection pool.""" + """Create database connection pool to active database.""" try: logger.info(f"Connecting to database at {self.config.host}:{self.config.port}") - self._pool = await asyncpg.create_pool( - dsn=self.config.dsn, - min_size=self.config.min_size, - max_size=self.config.max_size, - command_timeout=self.config.command_timeout, - server_settings={ - 'application_name': 'repologyapi', - 'search_path': 'repology,public', - } - ) + # Query which database is currently active + active_db = await self.query_active_database() + logger.info(f"Active database from control table: {active_db}") - logger.info("Database connection pool created successfully") + # Create pool to active database + self._pool = await self._create_pool(active_db) + self._current_database = active_db + + logger.info(f"Database connection pool created successfully for {active_db}") except Exception as e: logger.error(f"Failed to connect to database: {e}") raise + async def _refresh_connection(self, new_database: str) -> None: + """Switch connection pool to new active database.""" + old_database = self._current_database + old_pool = self._pool + + try: + logger.info(f"Refreshing connection pool: {old_database} → {new_database}") + + # Create new pool for new database + new_pool = await self._create_pool(new_database) + + # Atomic switch + self._pool = new_pool + self._current_database = new_database + + logger.info(f"✅ Switched to new database: {new_database}") + + # Gracefully close old pool (wait for active connections to finish) + if old_pool: + logger.info(f"Closing old pool for {old_database}...") + await old_pool.close() + logger.info(f"Old pool closed successfully") + + except Exception as e: + logger.error(f"Failed to refresh connection to {new_database}: {e}") + # Rollback if new pool creation failed + if old_pool and self._pool != old_pool: + logger.warning("Rolling back to old connection pool") + self._pool = old_pool + self._current_database = old_database + raise + + async def refresh_if_needed(self) -> bool: + """Check if active database changed and refresh connection if needed.""" + try: + active_db = await self.query_active_database() + + if active_db != self._current_database: + logger.info(f"Active database changed: {self._current_database} → {active_db}") + await self._refresh_connection(active_db) + return True + + return False + + except Exception as e: + logger.error(f"Error during refresh check: {e}") + # Don't disrupt service on refresh errors + return False + + async def _refresh_loop(self) -> None: + """Background task that periodically checks for database changes.""" + logger.info(f"Started database refresh loop (interval: {self.config.refresh_interval}s)") + + while not self._shutdown_event.is_set(): + try: + # Wait for either shutdown or refresh interval + try: + await asyncio.wait_for( + self._shutdown_event.wait(), + timeout=self.config.refresh_interval + ) + # Shutdown event was set + break + except asyncio.TimeoutError: + # Timeout = time to check for refresh + pass + + # Check if database needs refresh + changed = await self.refresh_if_needed() + if changed: + logger.info("Database rotation detected and applied") + + except Exception as e: + logger.error(f"Error in refresh loop: {e}") + # Continue running despite errors + await asyncio.sleep(5) # Brief pause before retry + + logger.info("Database refresh loop stopped") + + def start_refresh_task(self) -> None: + """Start background task to monitor for database changes.""" + if self._refresh_task is None or self._refresh_task.done(): + self._shutdown_event.clear() + self._refresh_task = asyncio.create_task(self._refresh_loop()) + logger.info("Database refresh task started") + else: + logger.warning("Refresh task already running") + + def stop_refresh_task(self) -> None: + """Stop background refresh task.""" + if self._refresh_task and not self._refresh_task.done(): + logger.info("Stopping database refresh task...") + self._shutdown_event.set() + else: + logger.info("Refresh task not running") + async def disconnect(self) -> None: """Close database connection pool.""" + # Stop refresh task first + self.stop_refresh_task() + if self._refresh_task: + try: + await asyncio.wait_for(self._refresh_task, timeout=5) + except asyncio.TimeoutError: + logger.warning("Refresh task did not stop in time") + + # Close connection pool if self._pool: logger.info("Closing database connection pool") await self._pool.close() self._pool = None + self._current_database = None @property def pool(self) -> asyncpg.Pool: @@ -94,10 +237,11 @@ async def health_check(self) -> bool: async def init_database(config: DatabaseConfig) -> None: - """Initialize global database connection.""" + """Initialize global database connection with periodic refresh.""" global _db_connection _db_connection = DatabaseConnection(config) await _db_connection.connect() + _db_connection.start_refresh_task() async def close_database() -> None: diff --git a/src/repologyapi/database/service.py b/src/repologyapi/database/service.py index 6885e18..0d040b9 100644 --- a/src/repologyapi/database/service.py +++ b/src/repologyapi/database/service.py @@ -41,7 +41,7 @@ async def get_connection_info(self) -> dict[str, str]: return { "host": config.host, "port": str(config.port), - "database": config.database, + "database": self.db_connection.database, "user": config.dbuser, "pool_min_size": str(config.min_size), "pool_max_size": str(config.max_size),