db/database.py

"""

PostgreSQL and SQLite database module for NHS Unified Nursing Validator.



Provides database connection pooling, schema management, and ORM operations

for chat history, users, sessions, and audit logs.

"""

import os
import json
import logging
import sqlite3
import threading
from datetime import datetime, date
from typing import Optional, List, Dict, Any, Generator, Tuple
from contextlib import contextmanager

try:
    import psycopg2
    from psycopg2 import pool, sql
    from psycopg2.extras import RealDictCursor, execute_values
    PSYCOPG2_AVAILABLE = True
except ImportError:
    PSYCOPG2_AVAILABLE = False
    # Define dummy classes to prevent NameError
    pool = None  
    RealDictCursor = None

from core.settings import settings
from core.safe_logging import mask_identifier, log_exception_safe

logger = logging.getLogger(__name__)

# Connection pool for Postgres (initialized on first use)
_pg_pool = None

# Thread-local storage for SQLite connections (since they can't be shared across threads easily)
_local_sqlite = threading.local()

def _get_sqlite_conn():
    """Get thread-local SQLite connection."""
    if not hasattr(_local_sqlite, "conn"):
        _local_sqlite.conn = sqlite3.connect(
            settings.SQLITE_DB_PATH, 
            check_same_thread=False,
            detect_types=sqlite3.PARSE_DECLTYPES
        )
        _local_sqlite.conn.row_factory = sqlite3.Row
    return _local_sqlite.conn

def init_connection_pool():
    """Initialize the PostgreSQL connection pool if needed."""
    global _pg_pool
    if settings.DB_TYPE == "postgres":
        if not PSYCOPG2_AVAILABLE:
            raise ImportError("psycopg2 is required for 'postgres' DB_TYPE")
            
        if _pg_pool is None:
            try:
                _pg_pool = psycopg2.pool.SimpleConnectionPool(
                    settings.DB_POOL_MIN,
                    settings.DB_POOL_MAX,
                    host=settings.DB_HOST,
                    port=settings.DB_PORT,
                    database=settings.DB_NAME,
                    user=settings.DB_USER,
                    password=settings.DB_PASSWORD,
                    connect_timeout=5,
                )
                logger.info("PostgreSQL connection pool initialized successfully")
            except Exception as e:
                log_exception_safe(logger, "Failed to initialize PG pooling", e)
                raise

@contextmanager
def get_connection():
    """Context manager for database connections (Postgres or SQLite)."""
    if settings.DB_TYPE == "sqlite":
        conn = _get_sqlite_conn()
        try:
            yield conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            log_exception_safe(logger, "SQLite DB error", e)
            raise
    else:
        # Postgres
        init_connection_pool()
        conn = _pg_pool.getconn()
        try:
            yield conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            log_exception_safe(logger, "Postgres DB error", e)
            raise
        finally:
            _pg_pool.putconn(conn)

def _dict_factory(cursor, row):
    """Custom dict factory for sqlite3 rows to match RealDictCursor behavior."""
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d

def _adapt_query(query: str) -> str:
    """Adapt SQL query for SQLite (replace %s with ?)."""
    if settings.DB_TYPE == "sqlite":
        return query.replace("%s", "?")
    return query

def _json_serialize(data: Any) -> Any:
    """Serialize data to JSON string if using SQLite, else return object for PG adapter."""
    if settings.DB_TYPE == "sqlite":
        return json.dumps(data) if data is not None else None
    return json.dumps(data) if data is not None else None # PG usually needs string unless using Json adapter

def init_database():
    """Initialize database schema and tables."""
    logger.info(f"Initializing database schema ({settings.DB_TYPE})...")
    with get_connection() as conn:
        cur = conn.cursor()
        try:
            # Type Abstractions
            if settings.DB_TYPE == "sqlite":
                PK_TYPE = "INTEGER PRIMARY KEY AUTOINCREMENT"
                JSON_TYPE = "TEXT"
                BOOL_TYPE = "BOOLEAN" # SQLite connects this to int, but declares fine
                TS_DEFAULT = "DEFAULT CURRENT_TIMESTAMP"
            else:
                PK_TYPE = "SERIAL PRIMARY KEY"
                JSON_TYPE = "JSONB"
                BOOL_TYPE = "BOOLEAN"
                TS_DEFAULT = "DEFAULT CURRENT_TIMESTAMP"

            # Create users table
            cur.execute(f"""

                CREATE TABLE IF NOT EXISTS users (

                    id {PK_TYPE},

                    username VARCHAR(50) UNIQUE NOT NULL,

                    password_hash VARCHAR(255) NOT NULL,

                    role VARCHAR(20) NOT NULL,

                    email VARCHAR(100),

                    created_at TIMESTAMP {TS_DEFAULT},

                    updated_at TIMESTAMP {TS_DEFAULT},

                    is_active {BOOL_TYPE} DEFAULT 1,

                    last_login TIMESTAMP

                )

            """)

            # Create sessions table
            cur.execute(f"""

                CREATE TABLE IF NOT EXISTS sessions (

                    id {PK_TYPE},

                    user_id INTEGER NOT NULL REFERENCES users(id),

                    session_token VARCHAR(255) UNIQUE NOT NULL,

                    created_at TIMESTAMP {TS_DEFAULT},

                    expires_at TIMESTAMP NOT NULL,

                    ip_address VARCHAR(45),

                    user_agent TEXT,

                    is_active {BOOL_TYPE} DEFAULT 1

                )

            """)

            # Create chat_history table
            cur.execute(f"""

                CREATE TABLE IF NOT EXISTS chat_history (

                    id {PK_TYPE},

                    user_id INTEGER NOT NULL REFERENCES users(id),

                    session_id INTEGER REFERENCES sessions(id),

                    role VARCHAR(20) NOT NULL,

                    content TEXT NOT NULL,

                    created_at TIMESTAMP {TS_DEFAULT},

                    metadata {JSON_TYPE}

                )

            """)

            # Create audit_logs table
            cur.execute(f"""

                CREATE TABLE IF NOT EXISTS audit_logs (

                    id {PK_TYPE},

                    user_id INTEGER REFERENCES users(id),

                    action VARCHAR(100) NOT NULL,

                    resource_type VARCHAR(50),

                    resource_id VARCHAR(100),

                    changes {JSON_TYPE},

                    created_at TIMESTAMP {TS_DEFAULT},

                    ip_address VARCHAR(45)

                )

            """)

            # Create analytics_events table
            cur.execute(f"""

                CREATE TABLE IF NOT EXISTS analytics_events (

                    id {PK_TYPE},

                    user_id INTEGER NOT NULL REFERENCES users(id),

                    event_type VARCHAR(50) NOT NULL,

                    event_name VARCHAR(100) NOT NULL,

                    data {JSON_TYPE},

                    created_at TIMESTAMP {TS_DEFAULT}

                )

            """)

            # Create Indexes
            if settings.DB_TYPE == "sqlite":
                 cur.execute("CREATE INDEX IF NOT EXISTS idx_chat_user_id ON chat_history(user_id)")
                 cur.execute("CREATE INDEX IF NOT EXISTS idx_analytics_user_id ON analytics_events(user_id)")
            else:
                 cur.execute("CREATE INDEX IF NOT EXISTS idx_chat_user_id ON chat_history(user_id)")
                 cur.execute("CREATE INDEX IF NOT EXISTS idx_analytics_user_id ON analytics_events(user_id)")

            conn.commit()
            logger.info("Database schema initialized successfully")
        except Exception as e:
            log_exception_safe(logger, "Failed to initialize schema", e)
            raise

def add_user(username: str, password_hash: str, role: str, email: Optional[str] = None) -> int:
    """Add a new user to the database."""
    with get_connection() as conn:
        cur = conn.cursor()
        try:
            query = _adapt_query("""

                INSERT INTO users (username, password_hash, role, email)

                VALUES (%s, %s, %s, %s)

            """)
            
            if settings.DB_TYPE == "postgres":
                 query += " RETURNING id"
                 cur.execute(query, (username, password_hash, role, email))
                 user_id = cur.fetchone()[0]
            else:
                 cur.execute(query, (username, password_hash, role, email))
                 user_id = cur.lastrowid
                 
            logger.info(f"User created: {mask_identifier(username, 'user')} (ID: {mask_identifier(str(user_id), 'id')})")
            return user_id
        except Exception as e:  # Catch IntegrityError equivalent
            logger.warning(
                "User creation failed (possible duplicate or constraint violation).",
                exc_info=True,
            )
            raise  # Re-raise for controller handling if needed
        except Exception as e: # Catch IntegrityError equivalent
            log_exception_safe(logger, "User creation failed", e, level="warning")
            raise # Re-raise for controller handling if needed

def get_user(username: str) -> Optional[Dict[str, Any]]:
    """Get user by username."""
    with get_connection() as conn:
        if settings.DB_TYPE == "postgres":
            cur = conn.cursor(cursor_factory=RealDictCursor)
        else:
            conn.row_factory = sqlite3.Row
            cur = conn.cursor()
            
        cur.execute(_adapt_query(
            "SELECT id, username, password_hash, role, email, is_active "
            "FROM users WHERE username = %s"
        ), (username,))
        
        row = cur.fetchone()
        if row:
             return dict(row)
        return None

def update_last_login(user_id: int) -> None:
    """Update user's last login timestamp."""
    with get_connection() as conn:
        cur = conn.cursor()
        cur.execute(_adapt_query(
            "UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = %s"
        ), (user_id,))

def save_chat_message(

    user_id: int,

    role: str,

    content: str,

    session_id: Optional[int] = None,

    metadata: Optional[Dict] = None,

) -> int:
    """Save a chat message to the database."""
    with get_connection() as conn:
        cur = conn.cursor()
        metadata_json = _json_serialize(metadata)
        
        query = _adapt_query("""

            INSERT INTO chat_history

            (user_id, session_id, role, content, metadata)

            VALUES (%s, %s, %s, %s, %s)

        """)
        
        if settings.DB_TYPE == "postgres":
            query += " RETURNING id"
            cur.execute(query, (user_id, session_id, role, content, metadata_json))
            message_id = cur.fetchone()[0]
        else:
            cur.execute(query, (user_id, session_id, role, content, metadata_json))
            message_id = cur.lastrowid
            
        return message_id

def get_chat_history(

    user_id: int, limit: int = 100, offset: int = 0

) -> List[Dict[str, Any]]:
    """Get chat history for a user."""
    with get_connection() as conn:
        if settings.DB_TYPE == "postgres":
            cur = conn.cursor(cursor_factory=RealDictCursor)
        else:
            conn.row_factory = _dict_factory # Use dict wrapper for easier JSON handling
            cur = conn.cursor()
            
        cur.execute(_adapt_query("""

            SELECT id, role, content, created_at, metadata

            FROM chat_history

            WHERE user_id = %s

            ORDER BY created_at DESC

            LIMIT %s OFFSET %s

        """), (user_id, limit, offset))
        
        rows = cur.fetchall()
        
        # SQLite JSON handling (deserialize text back to dict)
        if settings.DB_TYPE == "sqlite":
             result = []
             for row in rows:
                 r = dict(row)
                 if r.get("metadata") and isinstance(r["metadata"], str):
                     try:
                         r["metadata"] = json.loads(r["metadata"])
                     except: 
                         pass
                 result.append(r)
             return result
             
        # Postgres already returns Dict if RealDictCursor handles JSONB? 
        # Actually psycopg2 returns dict for JSONB automatically.
        return [dict(r) for r in rows]

def clear_chat_history(user_id: int) -> int:
    """Clear all chat history for a user."""
    with get_connection() as conn:
        cur = conn.cursor()
        cur.execute(_adapt_query(
            "DELETE FROM chat_history WHERE user_id = %s"
        ), (user_id,))
        return cur.rowcount

def log_audit_event(

    user_id: Optional[int],

    action: str,

    resource_type: Optional[str] = None,

    resource_id: Optional[str] = None,

    changes: Optional[Dict] = None,

    ip_address: Optional[str] = None,

) -> int:
    """Log an audit event."""
    with get_connection() as conn:
        cur = conn.cursor()
        changes_json = _json_serialize(changes)
        
        query = _adapt_query("""

            INSERT INTO audit_logs

            (user_id, action, resource_type, resource_id, changes, ip_address)

            VALUES (%s, %s, %s, %s, %s, %s)

        """)
        
        if settings.DB_TYPE == "postgres":
            query += " RETURNING id"
            cur.execute(query, (user_id, action, resource_type, resource_id, changes_json, ip_address))
            return cur.fetchone()[0]
        else:
            cur.execute(query, (user_id, action, resource_type, resource_id, changes_json, ip_address))
            return cur.lastrowid

def log_analytics_event(

    user_id: int, event_type: str, event_name: str, data: Optional[Dict] = None

) -> int:
    """Log an analytics event."""
    with get_connection() as conn:
        cur = conn.cursor()
        data_json = _json_serialize(data)
        
        query = _adapt_query("""

            INSERT INTO analytics_events

            (user_id, event_type, event_name, data)

            VALUES (%s, %s, %s, %s)

        """)
        
        if settings.DB_TYPE == "postgres":
             query += " RETURNING id"
             cur.execute(query, (user_id, event_type, event_name, data_json))
             return cur.fetchone()[0]
        else:
             cur.execute(query, (user_id, event_type, event_name, data_json))
             return cur.lastrowid


def get_analytics_summary(

    start_date: Optional[str] = None, end_date: Optional[str] = None

) -> Dict[str, Any]:
    """Get analytics summary for a date range."""
    with get_connection() as conn:
        if settings.DB_TYPE == "postgres":
             cur = conn.cursor(cursor_factory=RealDictCursor)
        else:
             conn.row_factory = sqlite3.Row
             cur = conn.cursor()
             
        query = _adapt_query("""

            SELECT

                COUNT(DISTINCT user_id) as unique_users,

                COUNT(*) as total_events,

                event_type,

                DATE(created_at) as event_date

            FROM analytics_events

            WHERE 1=1

        """)
        params = []

        if start_date:
            query += " AND created_at >= %s"
            params.append(start_date)

        if end_date:
            query += " AND created_at <= %s"
            params.append(end_date)

        query += " GROUP BY event_type, DATE(created_at) ORDER BY event_date DESC"

        cur.execute(query, params)
        rows = cur.fetchall()
        return [dict(r) for r in rows]

def get_top_users(limit: int = 5) -> List[Dict[str, Any]]:
    """Get top active users by event count."""
    with get_connection() as conn:
        if settings.DB_TYPE == "postgres":
             cur = conn.cursor(cursor_factory=RealDictCursor)
        else:
             conn.row_factory = sqlite3.Row
             cur = conn.cursor()
             
        # Join with users table to get usernames
        query = _adapt_query("""

            SELECT u.username, COUNT(a.id) as event_count

            FROM analytics_events a

            JOIN users u ON a.user_id = u.id

            GROUP BY u.username

            ORDER BY event_count DESC

            LIMIT %s

        """)
        
        cur.execute(query, (limit,))
        rows = cur.fetchall()
        return [dict(r) for r in rows]

def get_audit_logs(

    user_id: Optional[int] = None,

    start_date: Optional[str] = None,

    end_date: Optional[str] = None,

    limit: int = 100,

) -> List[Dict[str, Any]]:
    """Get audit logs with optional filtering."""
    with get_connection() as conn:
        if settings.DB_TYPE == "postgres":
             cur = conn.cursor(cursor_factory=RealDictCursor)
        else:
             conn.row_factory = _dict_factory
             cur = conn.cursor()
             
        query = _adapt_query("SELECT * FROM audit_logs WHERE 1=1")
        params = []

        if user_id:
            query += " AND user_id = %s"
            params.append(user_id)

        if start_date:
            query += " AND created_at >= %s"
            params.append(start_date)

        if end_date:
            query += " AND created_at <= %s"
            params.append(end_date)

        query += " ORDER BY created_at DESC LIMIT %s"
        params.append(limit)

        cur.execute(query, params)
        rows = cur.fetchall()
        
        # JSON handling for SQLite
        if settings.DB_TYPE == "sqlite":
             result = []
             for row in rows:
                 r = dict(row)
                 if r.get("changes") and isinstance(r["changes"], str):
                     try:
                         r["changes"] = json.loads(r["changes"])
                     except: pass
                 result.append(r)
             return result
             
        return [dict(r) for r in rows]

def close_connection_pool():
    """Close all connections in the pool."""
    global _pg_pool
    if _pg_pool:
        _pg_pool.closeall()
        _pg_pool = None