gandalf/db.py

"""Database operations for Gandalf network monitor."""
import json
import logging
import threading
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Optional

import pymysql
import pymysql.cursors

logger = logging.getLogger(__name__)

_config_cache = None
_local = threading.local()


def _config() -> dict:
    global _config_cache
    if _config_cache is None:
        with open('config.json') as f:
            _config_cache = json.load(f)['database']
    return _config_cache


def _new_conn(cfg: dict):
    return pymysql.connect(
        host=cfg['host'],
        port=cfg.get('port', 3306),
        user=cfg['user'],
        password=cfg['password'],
        database=cfg['name'],
        autocommit=True,
        cursorclass=pymysql.cursors.DictCursor,
        connect_timeout=10,
        charset='utf8mb4',
        # All app code treats database times as UTC. Without this, a server in another
        # time zone shifts displayed ages and timed suppression expiries by its offset.
        init_command="SET time_zone='+00:00'",
    )


@contextmanager
def get_conn():
    """Yield a per-thread cached database connection, reconnecting as needed."""
    cfg = _config()
    conn = getattr(_local, 'conn', None)
    if conn is None:
        conn = _new_conn(cfg)
        _local.conn = conn
    else:
        try:
            conn.ping(reconnect=True)
        except Exception:
            try:
                conn.close()
            except Exception:
                pass
            _local.conn = None
            conn = _new_conn(cfg)
            _local.conn = conn
    yield conn


# ---------------------------------------------------------------------------
# Monitor state (key/value store)
# ---------------------------------------------------------------------------

def set_state(key: str, value) -> None:
    if not isinstance(value, str):
        value = json.dumps(value, default=str)
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO monitor_state (key_name, value)
                   VALUES (%s, %s)
                   ON DUPLICATE KEY UPDATE value=VALUES(value), updated_at=NOW()""",
                (key, value),
            )


def get_state(key: str, default=None):
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT value FROM monitor_state WHERE key_name=%s', (key,))
            row = cur.fetchone()
            return row['value'] if row else default


# ---------------------------------------------------------------------------
# Interface baseline tracking
# ---------------------------------------------------------------------------

def get_baseline() -> dict:
    raw = get_state('interface_baseline')
    if raw:
        try:
            return json.loads(raw)
        except Exception:
            logger.error('Failed to parse interface_baseline JSON; resetting baseline')
    return {}


def set_baseline(baseline: dict) -> None:
    set_state('interface_baseline', json.dumps(baseline))


# ---------------------------------------------------------------------------
# Network events
# ---------------------------------------------------------------------------

def upsert_event(
    event_type: str,
    severity: str,
    source_type: str,
    target_name: str,
    target_detail: str,
    description: str,
) -> tuple:
    """Insert or update a network event. Returns (id, is_new, consecutive_failures, ticket_id)."""
    detail = target_detail or ''
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT id, consecutive_failures, ticket_id FROM network_events
                   WHERE event_type=%s AND target_name=%s AND target_detail=%s
                   AND resolved_at IS NULL LIMIT 1""",
                (event_type, target_name, detail),
            )
            existing = cur.fetchone()

            if existing:
                new_count = existing['consecutive_failures'] + 1
                cur.execute(
                    """UPDATE network_events
                       SET last_seen=NOW(), consecutive_failures=%s, description=%s
                       WHERE id=%s""",
                    (new_count, description, existing['id']),
                )
                return existing['id'], False, new_count, existing.get('ticket_id')
            else:
                cur.execute(
                    """INSERT INTO network_events
                       (event_type, severity, source_type, target_name, target_detail, description)
                       VALUES (%s, %s, %s, %s, %s, %s)""",
                    (event_type, severity, source_type, target_name, detail, description),
                )
                return cur.lastrowid, True, 1, None


def get_events_for_target(target_name: str, limit: int = 25) -> list:
    """Return a device's events, unresolved first, then most recent."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT * FROM network_events WHERE target_name=%s
                   ORDER BY resolved_at IS NOT NULL, COALESCE(resolved_at, last_seen) DESC
                   LIMIT %s""",
                (target_name, limit),
            )
            rows = cur.fetchall()
            for r in rows:
                for k in ('first_seen', 'last_seen', 'resolved_at'):
                    if r.get(k) and hasattr(r[k], 'isoformat'):
                        r[k] = r[k].isoformat() + 'Z'
            return rows


def resolve_event(event_type: str, target_name: str, target_detail: str = '') -> None:
    detail = target_detail or ''
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """UPDATE network_events SET resolved_at=NOW()
                   WHERE event_type=%s AND target_name=%s AND target_detail=%s
                   AND resolved_at IS NULL""",
                (event_type, target_name, detail),
            )


def get_open_event_targets(event_type: str) -> list:
    """(target_name, target_detail) of each unresolved event of one type."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                'SELECT target_name, target_detail FROM network_events WHERE event_type=%s AND resolved_at IS NULL',
                (event_type,),
            )
            return [(r['target_name'], r['target_detail']) for r in cur.fetchall()]


def set_ticket_id(event_id: int, ticket_id: str) -> None:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                'UPDATE network_events SET ticket_id=%s WHERE id=%s',
                (ticket_id, event_id),
            )


def get_active_events(limit: int = 200, offset: int = 0) -> list:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT * FROM network_events
                   WHERE resolved_at IS NULL
                   ORDER BY
                     FIELD(severity,'critical','warning','info'),
                     first_seen DESC
                   LIMIT %s OFFSET %s""",
                (limit, offset),
            )
            rows = cur.fetchall()
            for r in rows:
                for k in ('first_seen', 'last_seen'):
                    if r.get(k) and hasattr(r[k], 'isoformat'):
                        r[k] = r[k].isoformat() + 'Z'
            return rows


def count_active_events() -> int:
    """Return count of all unresolved events (for pagination)."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT COUNT(*) AS n FROM network_events WHERE resolved_at IS NULL"
            )
            return cur.fetchone()['n']


def get_recent_resolved(hours: int = 24, limit: int = 50) -> list:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT * FROM network_events
                   WHERE resolved_at IS NOT NULL
                   AND resolved_at > DATE_SUB(NOW(), INTERVAL %s HOUR)
                   ORDER BY resolved_at DESC LIMIT %s""",
                (hours, limit),
            )
            rows = cur.fetchall()
            for r in rows:
                for k in ('first_seen', 'last_seen', 'resolved_at'):
                    if r.get(k) and hasattr(r[k], 'isoformat'):
                        r[k] = r[k].isoformat() + 'Z'
            return rows


def get_status_summary() -> dict:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT severity, COUNT(*) as cnt FROM network_events
                   WHERE resolved_at IS NULL GROUP BY severity"""
            )
            counts = {r['severity']: r['cnt'] for r in cur.fetchall()}
            cur.execute(
                """SELECT COUNT(*) as cnt FROM network_events
                   WHERE resolved_at IS NOT NULL
                   AND resolved_at > DATE_SUB(NOW(), INTERVAL 24 HOUR)"""
            )
            resolved_24h = cur.fetchone()['cnt']
            return {
                'critical': counts.get('critical', 0),
                'warning': counts.get('warning', 0),
                'info': counts.get('info', 0),
                'resolved_24h': resolved_24h,
            }


# ---------------------------------------------------------------------------
# Suppression rules
# ---------------------------------------------------------------------------

def get_active_suppressions() -> list:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT * FROM suppression_rules
                   WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
                   ORDER BY created_at DESC"""
            )
            rows = cur.fetchall()
            for r in rows:
                for k in ('created_at', 'expires_at'):
                    if r.get(k) and hasattr(r[k], 'isoformat'):
                        r[k] = r[k].isoformat() + 'Z'
            return rows


def get_suppression_history(limit: int = 50) -> list:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                'SELECT * FROM suppression_rules ORDER BY created_at DESC LIMIT %s',
                (limit,),
            )
            rows = cur.fetchall()
            for r in rows:
                for k in ('created_at', 'expires_at'):
                    if r.get(k) and hasattr(r[k], 'isoformat'):
                        r[k] = r[k].isoformat() + 'Z'
            return rows


def create_suppression(
    target_type: str,
    target_name: str,
    target_detail: str,
    reason: str,
    suppressed_by: str,
    expires_minutes: Optional[int] = None,
) -> int:
    expires_at = None
    if expires_minutes:
        expires_at = datetime.now(timezone.utc) + timedelta(minutes=int(expires_minutes))
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO suppression_rules
                   (target_type, target_name, target_detail, reason, suppressed_by, expires_at, active)
                   VALUES (%s, %s, %s, %s, %s, %s, TRUE)""",
                (target_type, target_name or '', target_detail or '', reason, suppressed_by, expires_at),
            )
            return cur.lastrowid


def deactivate_suppression(sup_id: int) -> None:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                'UPDATE suppression_rules SET active=FALSE WHERE id=%s', (sup_id,)
            )


def cleanup_expired_suppressions() -> int:
    """Mark expired time-limited suppressions as inactive. Returns count deactivated."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """UPDATE suppression_rules
                   SET active=FALSE
                   WHERE active=TRUE AND expires_at IS NOT NULL AND expires_at <= NOW()"""
            )
            n = cur.rowcount
    if n:
        logger.info(f'Deactivated {n} expired suppression(s)')
    return n


def ensure_events_schema() -> None:
    """Add columns newer than the original events table, so deploys need no manual migration."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute("SHOW COLUMNS FROM network_events LIKE 'recovery_notified'")
            if cur.fetchone():
                return
            cur.execute(
                'ALTER TABLE network_events ADD COLUMN recovery_notified TINYINT(1) NOT NULL DEFAULT 0'
            )
            # Events that cleared before this existed are history, not owed a comment
            cur.execute(
                'UPDATE network_events SET recovery_notified=1, last_seen=last_seen WHERE resolved_at IS NOT NULL'
            )


def get_pending_recovery_notes(hours: int = 24, exclude_types: tuple = ()) -> list:
    """Resolved events with a ticket whose recovery comment hasn't been posted yet."""
    sql = """SELECT id, event_type, target_name, target_detail, ticket_id,
                    first_seen, resolved_at, consecutive_failures
             FROM network_events
             WHERE resolved_at IS NOT NULL AND ticket_id IS NOT NULL AND recovery_notified=0
             AND resolved_at > DATE_SUB(NOW(), INTERVAL %s HOUR)"""
    params: list = [hours]
    if exclude_types:
        sql += f' AND event_type NOT IN ({", ".join(["%s"] * len(exclude_types))})'  # nosec B608 - placeholders only
        params.extend(exclude_types)
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(sql + ' ORDER BY resolved_at', params)
            return list(cur.fetchall())


def mark_recovery_notified(event_id: int) -> None:
    with get_conn() as conn:
        with conn.cursor() as cur:
            # last_seen updates itself on any write; keep it as the last failing check
            cur.execute(
                'UPDATE network_events SET recovery_notified=1, last_seen=last_seen WHERE id=%s',
                (event_id,),
            )


def event_stats(event_type: str, target_name: str, target_detail: str = '') -> dict:
    """How often one alert has fired before, for a ticket's history section.

    Returns {'last_7d', 'last_30d', 'last_first_seen', 'last_resolved_at'} over resolved events.
    """
    key = (event_type, target_name, target_detail or '')
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT COALESCE(SUM(first_seen > DATE_SUB(NOW(), INTERVAL 7 DAY)), 0) AS last_7d,
                          COUNT(*) AS last_30d
                   FROM network_events
                   WHERE event_type=%s AND target_name=%s AND target_detail=%s
                   AND resolved_at IS NOT NULL AND first_seen > DATE_SUB(NOW(), INTERVAL 30 DAY)""",
                key,
            )
            counts = cur.fetchone() or {}
            cur.execute(
                """SELECT first_seen, resolved_at FROM network_events
                   WHERE event_type=%s AND target_name=%s AND target_detail=%s
                   AND resolved_at IS NOT NULL ORDER BY resolved_at DESC LIMIT 1""",
                key,
            )
            last = cur.fetchone() or {}
    return {
        'last_7d': int(counts.get('last_7d') or 0),
        'last_30d': int(counts.get('last_30d') or 0),
        'last_first_seen': last.get('first_seen'),
        'last_resolved_at': last.get('resolved_at'),
    }


def _history_where(since_hours: int, event_type: str = '', target: str = '',
                   severity: str = '', status: str = 'all') -> tuple:
    """WHERE clause shared by the history list and its summaries.

    An event is in range if it was open at any point in the window: it started
    in it, or it was still open when the window began.
    """
    clauses = ['(first_seen > DATE_SUB(NOW(), INTERVAL %s HOUR) OR '
               'resolved_at IS NULL OR resolved_at > DATE_SUB(NOW(), INTERVAL %s HOUR))']
    params: list = [since_hours, since_hours]
    if event_type:
        clauses.append('event_type=%s')
        params.append(event_type)
    if severity:
        clauses.append('severity=%s')
        params.append(severity)
    if target:
        like = '%' + target.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_') + '%'
        clauses.append('(target_name LIKE %s OR target_detail LIKE %s)')
        params.extend([like, like])
    if status == 'active':
        clauses.append('resolved_at IS NULL')
    elif status == 'resolved':
        clauses.append('resolved_at IS NOT NULL')
    return ' AND '.join(clauses), params


def _iso(rows: list, keys=('first_seen', 'last_seen', 'resolved_at')) -> list:
    for r in rows:
        for k in keys:
            if r.get(k) and hasattr(r[k], 'isoformat'):
                r[k] = r[k].isoformat() + 'Z'
    return rows


def get_events_history(since_hours: int, event_type: str = '', target: str = '', severity: str = '',
                       status: str = 'all', limit: int = 100, offset: int = 0) -> dict:
    """Events open at any time in the last `since_hours`, newest first, with totals.

    Returns {'events', 'total', 'by_type': {type: count}}.
    """
    where, params = _history_where(since_hours, event_type, target, severity, status)
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                f"""SELECT id, event_type, severity, source_type, target_name, target_detail, description,
                           first_seen, last_seen, resolved_at, consecutive_failures, ticket_id,
                           TIMESTAMPDIFF(SECOND, first_seen, COALESCE(resolved_at, NOW())) AS duration_s
                    FROM network_events WHERE {where}
                    ORDER BY first_seen DESC, id DESC LIMIT %s OFFSET %s""",  # nosec B608 - fixed clauses
                params + [limit, offset],
            )
            events = _iso(list(cur.fetchall()))
            cur.execute(
                f"""SELECT event_type, COUNT(*) AS n
                    FROM network_events WHERE {where} GROUP BY event_type ORDER BY n DESC""",  # nosec B608
                params,
            )
            groups = cur.fetchall()
    return {
        'events': events,
        'total': sum(int(g['n']) for g in groups),
        'by_type': {g['event_type']: int(g['n']) for g in groups},
    }


def get_flappers(since_hours: int, event_type: str = '', target: str = '', severity: str = '',
                 min_count: int = 2, limit: int = 10) -> list:
    """The alerts that fired most often in the window: the ones worth a root cause."""
    where, params = _history_where(since_hours, event_type, target, severity)
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                f"""SELECT event_type, target_name, target_detail, COUNT(*) AS n,
                           SUM(TIMESTAMPDIFF(SECOND, first_seen, COALESCE(resolved_at, NOW()))) AS total_s,
                           MAX(first_seen) AS last_seen, SUM(resolved_at IS NULL) AS open_now
                    FROM network_events WHERE {where}
                    GROUP BY event_type, target_name, target_detail
                    HAVING n >= %s ORDER BY n DESC, last_seen DESC LIMIT %s""",  # nosec B608 - fixed clauses
                params + [min_count, limit],
            )
            rows = list(cur.fetchall())
    for r in rows:
        r['n'], r['total_s'], r['open_now'] = int(r['n']), int(r['total_s'] or 0), int(r['open_now'] or 0)
    return _iso(rows, ('last_seen',))


def get_event_types() -> list:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT DISTINCT event_type FROM network_events ORDER BY event_type')
            return [r['event_type'] for r in cur.fetchall()]


def purge_old_resolved_events(days: int = 90) -> int:
    """Delete resolved events older than `days` days. Returns count deleted."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """DELETE FROM network_events
                   WHERE resolved_at IS NOT NULL
                   AND resolved_at < DATE_SUB(NOW(), INTERVAL %s DAY)""",
                (days,),
            )
            n = cur.rowcount
    if n:
        logger.info(f'Purged {n} old resolved event(s) (>{days}d)')
    return n


# ---------------------------------------------------------------------------
# WAN modem optical history
# ---------------------------------------------------------------------------

def ensure_modem_schema() -> None:
    """Create the modem history table so deploys need no manual migration."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """CREATE TABLE IF NOT EXISTS modem_samples (
                     sampled_at   TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP PRIMARY KEY,
                     rx_power_dbm DECIMAL(5,1) NULL,
                     tx_power_dbm DECIMAL(5,1) NULL,
                     temp_c       DECIMAL(5,1) NULL
                   ) ENGINE=InnoDB"""
            )


def record_modem_sample(rx_power_dbm, tx_power_dbm, temp_c) -> None:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO modem_samples (sampled_at, rx_power_dbm, tx_power_dbm, temp_c)
                   VALUES (NOW(), %s, %s, %s)
                   ON DUPLICATE KEY UPDATE rx_power_dbm=VALUES(rx_power_dbm),
                     tx_power_dbm=VALUES(tx_power_dbm), temp_c=VALUES(temp_c)""",
                (rx_power_dbm, tx_power_dbm, temp_c),
            )


def get_modem_rx_values(days: int = 7) -> list:
    """Return recent Rx power readings (dBm) for computing a drift baseline."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT rx_power_dbm FROM modem_samples
                   WHERE rx_power_dbm IS NOT NULL
                   AND sampled_at > DATE_SUB(NOW(), INTERVAL %s DAY)""",
                (days,),
            )
            return [float(r['rx_power_dbm']) for r in cur.fetchall()]


def get_modem_history(hours: int = 168, bucket_seconds: int = 3600) -> list:
    """Return optical history averaged into time buckets, oldest first."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT FLOOR(UNIX_TIMESTAMP(sampled_at) / %s) * %s AS ts,
                          ROUND(AVG(rx_power_dbm), 1) AS rx_avg,
                          MIN(rx_power_dbm) AS rx_min,
                          ROUND(AVG(tx_power_dbm), 1) AS tx_avg,
                          ROUND(AVG(temp_c), 1) AS temp_avg
                   FROM modem_samples
                   WHERE sampled_at > DATE_SUB(NOW(), INTERVAL %s HOUR)
                   GROUP BY ts ORDER BY ts""",
                (bucket_seconds, bucket_seconds, hours),
            )
            return [
                {k: (float(v) if v is not None else None) for k, v in row.items()}
                for row in cur.fetchall()
            ]


def purge_old_modem_samples(days: int = 90) -> int:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                'DELETE FROM modem_samples WHERE sampled_at < DATE_SUB(NOW(), INTERVAL %s DAY)',
                (days,),
            )
            return cur.rowcount


def check_suppressed(suppressions: list, target_type: str, target_name: str, target_detail: str = '') -> bool:
    """Check suppression against a pre-loaded list (avoids per-call DB queries)."""
    for s in suppressions:
        if s['target_type'] == 'all':
            return True
        if s['target_type'] == target_type and s['target_name'] == target_name:
            if not (s.get('target_detail') or ''):
                return True
            if target_detail and s.get('target_detail') == target_detail:
                return True
    return False


def is_suppressed(target_type: str, target_name: str, target_detail: str = '') -> bool:
    with get_conn() as conn:
        with conn.cursor() as cur:
            # Global suppression (all)
            cur.execute(
                """SELECT id FROM suppression_rules
                   WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
                   AND target_type='all' LIMIT 1"""
            )
            if cur.fetchone():
                return True

            if not target_name:
                return False

            # Host-level suppression (covers all interfaces on that host)
            cur.execute(
                """SELECT id FROM suppression_rules
                   WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
                   AND target_type=%s AND target_name=%s
                   AND target_detail='' LIMIT 1""",
                (target_type, target_name),
            )
            if cur.fetchone():
                return True

            # Interface/device-specific suppression
            if target_detail:
                cur.execute(
                    """SELECT id FROM suppression_rules
                       WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
                       AND target_type=%s AND target_name=%s AND target_detail=%s LIMIT 1""",
                    (target_type, target_name, target_detail),
                )
                if cur.fetchone():
                    return True

            return False


# ---------------------------------------------------------------------------
# Cutsheets (see cutsheet.py). Every change writes a cutsheet_history row in
# the same transaction, so the log can't miss an edit.
# ---------------------------------------------------------------------------

_CUTSHEET_SCHEMA = (
    """CREATE TABLE IF NOT EXISTS cutsheet_items (
         id          INT AUTO_INCREMENT PRIMARY KEY,
         name        VARCHAR(100) NOT NULL,
         kind        VARCHAR(20)  NOT NULL,
         port_count  SMALLINT     NULL,
         location    VARCHAR(255) NOT NULL DEFAULT '',
         rack_u      SMALLINT     NULL,
         rack_height SMALLINT     NULL,
         rack        VARCHAR(100) NOT NULL DEFAULT '',
         shelf       VARCHAR(100) NOT NULL DEFAULT '',
         device      VARCHAR(255) NOT NULL DEFAULT '',
         mount       VARCHAR(10)  NOT NULL DEFAULT 'front',
         notes       TEXT         NULL,
         created_at  TIMESTAMP    DEFAULT CURRENT_TIMESTAMP,
         updated_at  TIMESTAMP    DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
         UNIQUE KEY uq_item_name (name)
       ) ENGINE=InnoDB""",
    """CREATE TABLE IF NOT EXISTS cutsheet_cables (
         id            INT AUTO_INCREMENT PRIMARY KEY,
         label         VARCHAR(32)  NULL,
         a_device      VARCHAR(100) NOT NULL,
         a_port        VARCHAR(64)  NOT NULL DEFAULT '',
         a_side        VARCHAR(5)   NOT NULL DEFAULT '',
         b_device      VARCHAR(100) NOT NULL DEFAULT '',
         b_port        VARCHAR(64)  NOT NULL DEFAULT '',
         b_side        VARCHAR(5)   NOT NULL DEFAULT '',
         cable_type    VARCHAR(32)  NOT NULL DEFAULT '',
         length_m      DECIMAL(5,2) NULL,
         color         VARCHAR(32)  NOT NULL DEFAULT '',
         status        VARCHAR(10)  NOT NULL DEFAULT 'installed',
         label_applied BOOLEAN      NOT NULL DEFAULT FALSE,
         notes         TEXT         NULL,
         created_by    VARCHAR(255) NOT NULL DEFAULT '',
         created_at    TIMESTAMP    DEFAULT CURRENT_TIMESTAMP,
         updated_by    VARCHAR(255) NOT NULL DEFAULT '',
         updated_at    TIMESTAMP    DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
         UNIQUE KEY uq_cable_label (label),
         INDEX idx_cable_a (a_device, a_port),
         INDEX idx_cable_b (b_device, b_port)
       ) ENGINE=InnoDB""",
    """CREATE TABLE IF NOT EXISTS cutsheet_port_labels (
         item_id INT         NOT NULL,
         port    VARCHAR(64) NOT NULL,
         label   VARCHAR(64) NOT NULL DEFAULT '',
         applied BOOLEAN     NOT NULL DEFAULT FALSE,
         PRIMARY KEY (item_id, port)
       ) ENGINE=InnoDB""",
    """CREATE TABLE IF NOT EXISTS cutsheet_history (
         id          INT AUTO_INCREMENT PRIMARY KEY,
         changed_at  TIMESTAMP    DEFAULT CURRENT_TIMESTAMP,
         changed_by  VARCHAR(255) NOT NULL,
         entity      VARCHAR(12)  NOT NULL,
         entity_id   INT          NOT NULL,
         action      VARCHAR(10)  NOT NULL,
         summary     VARCHAR(500) NOT NULL,
         before_json TEXT         NULL,
         after_json  TEXT         NULL,
         INDEX idx_hist_time (changed_at),
         INDEX idx_hist_entity (entity, entity_id)
       ) ENGINE=InnoDB""",
    """CREATE TABLE IF NOT EXISTS cutsheet_racks (
         id         INT AUTO_INCREMENT PRIMARY KEY,
         name       VARCHAR(100) NOT NULL,
         units      SMALLINT     NOT NULL,
         notes      TEXT         NULL,
         created_at TIMESTAMP    DEFAULT CURRENT_TIMESTAMP,
         updated_at TIMESTAMP    DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
         UNIQUE KEY uq_rack_name (name)
       ) ENGINE=InnoDB""",
    # Where a device lives in the asset inventory (HomeBox), set by hand on its Gandalf page
    """CREATE TABLE IF NOT EXISTS device_links (
         name          VARCHAR(255) NOT NULL PRIMARY KEY,
         inventory_url VARCHAR(512) NOT NULL,
         updated_by    VARCHAR(255) NOT NULL,
         updated_at    TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
       ) ENGINE=InnoDB""",
)


def get_device_link(name: str) -> str:
    """The device's inventory URL, or ''."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT inventory_url FROM device_links WHERE name=%s', (name,))
            row = cur.fetchone()
    return row['inventory_url'] if row else ''


def set_device_link(name: str, inventory_url: str, user: str) -> None:
    """Save a device's inventory URL; an empty URL removes it."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            if not inventory_url:
                cur.execute('DELETE FROM device_links WHERE name=%s', (name,))
                return
            cur.execute(
                """INSERT INTO device_links (name, inventory_url, updated_by) VALUES (%s, %s, %s)
                   ON DUPLICATE KEY UPDATE inventory_url=VALUES(inventory_url), updated_by=VALUES(updated_by)""",
                (name, inventory_url, user),
            )

_ITEM_FIELDS = ('name', 'kind', 'port_count', 'location', 'rack_u', 'rack_height', 'rack', 'shelf', 'device',
                'mount', 'notes')
# Columns added to cutsheet_items after its first release: (name, definition)
_ITEM_COLUMNS_ADDED = (
    ('rack', "VARCHAR(100) NOT NULL DEFAULT ''"),
    ('shelf', "VARCHAR(100) NOT NULL DEFAULT ''"),
    # The Gandalf device (UniFi or host name) an item is, when its own name differs
    ('device', "VARCHAR(255) NOT NULL DEFAULT ''"),
    ('mount', "VARCHAR(10) NOT NULL DEFAULT 'front'"),
)
_CABLE_FIELDS = ('label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side',
                 'cable_type', 'length_m', 'color', 'status', 'label_applied', 'notes')


class CutsheetConflict(Exception):
    """A change that would clash with existing records (duplicate name or label, item in use)."""


def ensure_cutsheet_schema() -> None:
    with get_conn() as conn:
        with conn.cursor() as cur:
            for statement in _CUTSHEET_SCHEMA:
                cur.execute(statement)
            cur.execute('SHOW COLUMNS FROM cutsheet_items')
            have = {r['Field'] for r in cur.fetchall()}
            for column, definition in _ITEM_COLUMNS_ADDED:
                if column not in have:
                    # Fixed names from _ITEM_COLUMNS_ADDED, never user input
                    cur.execute(f'ALTER TABLE cutsheet_items ADD COLUMN {column} {definition}')  # nosec B608


def _cs_jsonable(row: Optional[dict]) -> Optional[dict]:
    if row is None:
        return None
    out = {}
    for k, v in row.items():
        if hasattr(v, 'isoformat'):
            v = v.isoformat() + 'Z'
        elif v is not None and k == 'length_m':
            v = float(v)
        elif k in ('label_applied', 'applied'):
            v = bool(v)
        out[k] = v
    return out


def _cs_log(cur, user: str, entity: str, entity_id: int, action: str, summary: str,
            before: Optional[dict], after: Optional[dict]) -> None:
    cur.execute(
        """INSERT INTO cutsheet_history (changed_by, entity, entity_id, action, summary, before_json, after_json)
           VALUES (%s, %s, %s, %s, %s, %s, %s)""",
        (user, entity, entity_id, action, summary[:500],
         json.dumps(_cs_jsonable(before)) if before else None,
         json.dumps(_cs_jsonable(after)) if after else None),
    )


@contextmanager
def _cs_transaction():
    with get_conn() as conn:
        conn.begin()
        try:
            with conn.cursor() as cur:
                yield cur
            conn.commit()
        except pymysql.err.IntegrityError as e:
            conn.rollback()
            raise CutsheetConflict(_cs_integrity_message(e)) from e
        except Exception:
            conn.rollback()
            raise


def _cs_integrity_message(e: Exception) -> str:
    text = str(e)
    if 'uq_cable_label' in text:
        return 'Another cable already has that label'
    if 'uq_item_name' in text:
        return 'Another item already has that name'
    return 'That change conflicts with an existing record'


def cutsheet_load() -> dict:
    """Every cutsheet record, for one page render or API response."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT * FROM cutsheet_items ORDER BY name')
            items = [_cs_jsonable(r) for r in cur.fetchall()]
            cur.execute('SELECT * FROM cutsheet_cables ORDER BY id')
            cables = [_cs_jsonable(r) for r in cur.fetchall()]
            cur.execute(
                """SELECT l.item_id, i.name AS item_name, l.port, l.label, l.applied
                   FROM cutsheet_port_labels l JOIN cutsheet_items i ON i.id = l.item_id"""
            )
            port_labels = [_cs_jsonable(r) for r in cur.fetchall()]
    return {'items': items, 'cables': cables, 'port_labels': port_labels}


def cutsheet_history(limit: int = 100) -> list:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT id, changed_at, changed_by, entity, entity_id, action, summary
                   FROM cutsheet_history ORDER BY id DESC LIMIT %s""",
                (limit,),
            )
            return [_cs_jsonable(r) for r in cur.fetchall()]


def _cs_get(cur, table: str, row_id: int) -> Optional[dict]:
    cur.execute(f'SELECT * FROM {table} WHERE id=%s FOR UPDATE', (row_id,))  # nosec B608 (fixed names)
    return cur.fetchone()


def cutsheet_create_item(item: dict, user: str) -> int:
    with _cs_transaction() as cur:
        cur.execute(
            f'INSERT INTO cutsheet_items ({", ".join(_ITEM_FIELDS)}) '  # nosec B608 (fixed names)
            f'VALUES ({", ".join(["%s"] * len(_ITEM_FIELDS))})',
            tuple(item[f] for f in _ITEM_FIELDS),
        )
        item_id = cur.lastrowid
        _cs_log(cur, user, 'item', item_id, 'create', f'Added {item["kind"]} {item["name"]}', None, item)
        return item_id


def cutsheet_update_item(item_id: int, item: dict, user: str) -> bool:
    """Update an item. A rename carries over to the cables that reference it."""
    with _cs_transaction() as cur:
        before = _cs_get(cur, 'cutsheet_items', item_id)
        if before is None:
            return False
        cur.execute(
            f'UPDATE cutsheet_items SET {", ".join(f"{f}=%s" for f in _ITEM_FIELDS)} WHERE id=%s',  # nosec B608
            tuple(item[f] for f in _ITEM_FIELDS) + (item_id,),
        )
        summary = f'Edited {item["name"]}'
        if before['kind'] == 'shelf' and item['kind'] != 'shelf':
            on_it = _cs_on_shelf(cur, before['name'])
            if on_it:
                raise CutsheetConflict(f'{before["name"]} still holds {on_it} item(s); move them off the shelf first')
        if before['name'] != item['name']:
            for end in ('a', 'b'):
                cur.execute(
                    f'UPDATE cutsheet_cables SET {end}_device=%s, updated_by=%s WHERE {end}_device=%s',  # nosec B608
                    (item['name'], user, before['name']),
                )
            cur.execute('UPDATE cutsheet_items SET shelf=%s WHERE shelf=%s', (item['name'], before['name']))
            summary = f'Renamed {before["name"]} to {item["name"]}'
        _cs_log(cur, user, 'item', item_id, 'update', summary, before, item)
        return True


def _cs_on_shelf(cur, shelf: str) -> int:
    cur.execute('SELECT COUNT(*) AS n FROM cutsheet_items WHERE shelf=%s', (shelf,))
    return cur.fetchone()['n']


def cutsheet_delete_item(item_id: int, user: str) -> bool:
    with _cs_transaction() as cur:
        before = _cs_get(cur, 'cutsheet_items', item_id)
        if before is None:
            return False
        cur.execute(
            'SELECT COUNT(*) AS n FROM cutsheet_cables WHERE a_device=%s OR b_device=%s',
            (before['name'], before['name']),
        )
        used = cur.fetchone()['n']
        if used:
            raise CutsheetConflict(f'{before["name"]} still has {used} cable(s); move or delete them first')
        on_it = _cs_on_shelf(cur, before['name'])
        if on_it:
            raise CutsheetConflict(f'{before["name"]} still holds {on_it} item(s); move them off the shelf first')
        cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s', (item_id,))
        cur.execute('DELETE FROM cutsheet_items WHERE id=%s', (item_id,))
        _cs_log(cur, user, 'item', item_id, 'delete', f'Removed {before["name"]}', before, None)
        return True


def _cable_summary(c: dict) -> str:
    def end(device: str, port: str) -> str:
        return ' '.join(filter(None, [device or 'unterminated', port]))
    ends = f'{end(c["a_device"], c["a_port"])} ↔ {end(c["b_device"], c["b_port"])}'
    return f'{c.get("label") or "unlabeled cable"}: {ends}'


def cutsheet_create_cable(cable: dict, user: str) -> int:
    with _cs_transaction() as cur:
        cur.execute(
            f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) '  # nosec B608
            f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)',
            tuple(cable[f] for f in _CABLE_FIELDS) + (user, user),
        )
        cable_id = cur.lastrowid
        _cs_log(cur, user, 'cable', cable_id, 'create', f'Added {_cable_summary(cable)}', None, cable)
        return cable_id


def cutsheet_update_cable(cable_id: int, cable: dict, user: str) -> bool:
    with _cs_transaction() as cur:
        before = _cs_get(cur, 'cutsheet_cables', cable_id)
        if before is None:
            return False
        cur.execute(
            f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, updated_by=%s '  # nosec B608
            'WHERE id=%s',
            tuple(cable[f] for f in _CABLE_FIELDS) + (user, cable_id),
        )
        _cs_log(cur, user, 'cable', cable_id, 'update', f'Edited {_cable_summary(cable)}', before, cable)
        return True


def cutsheet_delete_cable(cable_id: int, user: str) -> bool:
    with _cs_transaction() as cur:
        before = _cs_get(cur, 'cutsheet_cables', cable_id)
        if before is None:
            return False
        cur.execute('DELETE FROM cutsheet_cables WHERE id=%s', (cable_id,))
        _cs_log(cur, user, 'cable', cable_id, 'delete', f'Deleted {_cable_summary(before)}', before, None)
        return True


def cutsheet_set_port_label(item_id: int, port: str, label: str, applied: bool, user: str) -> bool:
    with _cs_transaction() as cur:
        item = _cs_get(cur, 'cutsheet_items', item_id)
        if item is None:
            return False
        cur.execute('SELECT * FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port))
        before = cur.fetchone()
        if label:
            cur.execute(
                """INSERT INTO cutsheet_port_labels (item_id, port, label, applied) VALUES (%s, %s, %s, %s)
                   ON DUPLICATE KEY UPDATE label=VALUES(label), applied=VALUES(applied)""",
                (item_id, port, label, applied),
            )
        else:
            cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port))
        after = {'item_id': item_id, 'port': port, 'label': label, 'applied': applied} if label else None
        _cs_log(cur, user, 'port_label', item_id, 'update',
                f'{item["name"]} port {port} label: {label or "(cleared)"}', before, after)
        return True


def cutsheet_import_cables(cables: list, user: str) -> dict:
    """Add or update cables from a CSV import, matching on label. All or nothing."""
    added = updated = 0
    with _cs_transaction() as cur:
        for cable in cables:
            before = None
            if cable['label']:
                cur.execute('SELECT * FROM cutsheet_cables WHERE label=%s FOR UPDATE', (cable['label'],))
                before = cur.fetchone()
            if before:
                cur.execute(
                    f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, '  # nosec B608
                    'updated_by=%s WHERE id=%s',
                    tuple(cable[f] for f in _CABLE_FIELDS) + (user, before['id']),
                )
                _cs_log(cur, user, 'cable', before['id'], 'update', f'Imported {_cable_summary(cable)}',
                        before, cable)
                updated += 1
            else:
                cur.execute(
                    f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) '  # nosec
                    f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)',
                    tuple(cable[f] for f in _CABLE_FIELDS) + (user, user),
                )
                _cs_log(cur, user, 'cable', cur.lastrowid, 'create', f'Imported {_cable_summary(cable)}',
                        None, cable)
                added += 1
    return {'added': added, 'updated': updated}


# ---------------------------------------------------------------------------
# Racks (cutsheet_racks): name and height, edited on the rack page
# ---------------------------------------------------------------------------

def cutsheet_racks() -> list:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT id, name, units, notes FROM cutsheet_racks ORDER BY id')
            return list(cur.fetchall())


def cutsheet_save_rack(rack_id: Optional[int], rack: dict, user: str, previous: str = '') -> int:
    """Create (rack_id None) or update a rack. A rename carries over to the items in it.

    `previous` names the rack being replaced when the first rack is saved for the first
    time (it existed only as the default until then), so its items follow the new name.
    """
    with _cs_transaction() as cur:
        before = None
        if rack_id is not None:
            cur.execute('SELECT * FROM cutsheet_racks WHERE id=%s', (rack_id,))
            before = cur.fetchone()
            if before is None:
                raise CutsheetConflict('No such rack')
        cur.execute('SELECT name FROM cutsheet_racks WHERE LOWER(name)=LOWER(%s) AND id<>%s',
                    (rack['name'], rack_id or 0))
        clash = cur.fetchone()
        if clash:
            raise CutsheetConflict(f'A rack named {clash["name"]} already exists')
        if before is None:
            cur.execute('INSERT INTO cutsheet_racks (name, units, notes) VALUES (%s, %s, %s)',
                        (rack['name'], rack['units'], rack['notes']))
            rack_id = cur.lastrowid
            old_name = previous
        else:
            cur.execute('UPDATE cutsheet_racks SET name=%s, units=%s, notes=%s WHERE id=%s',
                        (rack['name'], rack['units'], rack['notes'], rack_id))
            old_name = before['name']
        if old_name and old_name != rack['name']:
            cur.execute('UPDATE cutsheet_items SET rack=%s WHERE rack=%s', (rack['name'], old_name))
        renamed = f' (was {old_name})' if old_name and old_name != rack['name'] else ''
        _cs_log(cur, user, 'rack', rack_id, 'update' if before else 'create',
                f'Rack {rack["name"]}: {rack["units"]}U{renamed}', _cs_jsonable(before), rack)
        return rack_id


def cutsheet_delete_rack(rack_id: int, user: str) -> bool:
    with _cs_transaction() as cur:
        cur.execute('SELECT * FROM cutsheet_racks WHERE id=%s', (rack_id,))
        before = cur.fetchone()
        if before is None:
            return False
        cur.execute('SELECT COUNT(*) AS n FROM cutsheet_items WHERE rack=%s', (before['name'],))
        n = cur.fetchone()['n']
        if n:
            raise CutsheetConflict(f'{before["name"]} still holds {n} item(s); move them to another rack first')
        cur.execute('DELETE FROM cutsheet_racks WHERE id=%s', (rack_id,))
        _cs_log(cur, user, 'rack', rack_id, 'delete', f'Removed rack {before["name"]}', _cs_jsonable(before), None)
        return True