#!/usr/bin/env python3
"""
Etherpad MySQL dry-run analyzer.

Read-only:
- connects to MySQL,
- inspects the Etherpad `store` table,
- counts key prefixes,
- enumerates pads,
- searches for ep_mypads-related metadata,
- writes a JSON report,
- performs no INSERT/UPDATE/DELETE operations.

Dependency:
    pip install mysql-connector-python
"""

from __future__ import annotations

import argparse
import json
import logging
import re
import sys
from collections import Counter
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any

import mysql.connector
from mysql.connector import Error as MySQLError


LOG = logging.getLogger("etherpad-dry-run")

PAD_BASE_RE = re.compile(r"^pad:(.+)$")
PAD_CHILD_MARKERS = (
    ":revs:",
    ":chat:",
    ":readonly:",
)


@dataclass
class PadInfo:
    pad_id: str
    head: int | None
    text_preview: str | None
    has_mypads_reference: bool


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Read-only analysis of an Etherpad MySQL database."
    )
    parser.add_argument("--host", required=True, help="MySQL host")
    parser.add_argument("--port", type=int, default=3306, help="MySQL port")
    parser.add_argument("--user", required=True, help="MySQL user")
    parser.add_argument("--password", required=True, help="MySQL password")
    parser.add_argument("--database", required=True, help="MySQL database name")
    parser.add_argument("--table", default="store", help="Etherpad store table")
    parser.add_argument(
        "--output",
        default="etherpad-dry-run-report.json",
        help="Path to the generated JSON report.",
    )
    parser.add_argument(
        "--sample-limit",
        type=int,
        default=30,
        help="Maximum number of ep_mypads-related records included in the report.",
    )
    parser.add_argument(
        "--pad-limit",
        type=int,
        default=0,
        help="Maximum number of pads to inspect; 0 means all.",
    )
    parser.add_argument("-v", "--verbose", action="store_true")
    return parser.parse_args()


def validate_identifier(identifier: str, label: str) -> str:
    if not re.fullmatch(r"[A-Za-z0-9_]+", identifier):
        raise ValueError(f"Unsafe {label}: {identifier!r}")
    return identifier


def decode_json(value: str) -> Any:
    try:
        return json.loads(value)
    except (TypeError, json.JSONDecodeError):
        return None


def extract_pad_text(value: str) -> tuple[int | None, str | None]:
    """
    Extracts only the current atext preview already present in pad:<id>.
    It does not reconstruct historical changesets.
    """
    obj = decode_json(value)
    if not isinstance(obj, dict):
        return None, None

    head = obj.get("head")
    if not isinstance(head, int):
        head = None

    text: str | None = None
    atext = obj.get("atext")
    if isinstance(atext, dict) and isinstance(atext.get("text"), str):
        text = atext["text"]
    elif isinstance(obj.get("text"), str):
        text = obj["text"]

    if text is not None:
        text = text.replace("\r", "")
        text = text[:300]

    return head, text


def query_all(cursor, sql: str, params: tuple[Any, ...] = ()) -> list[tuple]:
    cursor.execute(sql, params)
    return list(cursor.fetchall())


def main() -> int:
    args = parse_args()
    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(levelname)s: %(message)s",
    )

    try:
        table = validate_identifier(args.table, "table name")
        database = validate_identifier(args.database, "database name")
    except ValueError as exc:
        LOG.error("%s", exc)
        return 2

    connection = None
    try:
        LOG.info("Connecting to %s:%s/%s", args.host, args.port, database)
        connection = mysql.connector.connect(
            host=args.host,
            port=args.port,
            user=args.user,
            password=args.password or "",
            database=database,
            charset="utf8mb4",
            use_unicode=True,
            autocommit=False,
            connection_timeout=10,
        )

        # Explicit read-only transaction. No write statements are issued.
        connection.start_transaction(readonly=True, consistent_snapshot=True)
        cursor = connection.cursor()

        cursor.execute(
            """
            SELECT COUNT(*)
            FROM information_schema.tables
            WHERE table_schema = %s AND table_name = %s
            """,
            (database, table),
        )
        if cursor.fetchone()[0] != 1:
            LOG.error("Table `%s`.`%s` does not exist", database, table)
            return 3

        quoted_table = f"`{table}`"

        LOG.info("Reading key statistics")
        total_records = query_all(cursor, f"SELECT COUNT(*) FROM {quoted_table}")[0][0]

        prefix_rows = query_all(
            cursor,
            f"""
            SELECT SUBSTRING_INDEX(`key`, ':', 1) AS prefix, COUNT(*) AS amount
            FROM {quoted_table}
            GROUP BY prefix
            ORDER BY amount DESC, prefix ASC
            """,
        )
        prefix_counts = {str(prefix): int(amount) for prefix, amount in prefix_rows}

        LOG.info("Finding base pad records")
        pad_rows = query_all(
            cursor,
            f"""
            SELECT `key`, `value`
            FROM {quoted_table}
            WHERE `key` LIKE 'pad:%%'
            ORDER BY `key`
            """,
        )

        base_pad_rows: list[tuple[str, str]] = []
        for key, value in pad_rows:
            if any(marker in key for marker in PAD_CHILD_MARKERS):
                continue
            match = PAD_BASE_RE.match(key)
            if match:
                base_pad_rows.append((key, value))

        if args.pad_limit > 0:
            base_pad_rows = base_pad_rows[: args.pad_limit]

        LOG.info("Searching for ep_mypads metadata")
        mypads_rows = query_all(
            cursor,
            f"""
            SELECT `key`, `value`
            FROM {quoted_table}
            WHERE LOWER(`key`) LIKE '%%mypads%%'
               OR LOWER(`key`) LIKE '%%folder%%'
               OR LOWER(`key`) LIKE '%%workspace%%'
               OR LOWER(`key`) LIKE '%%userpads%%'
               OR LOWER(`value`) LIKE '%%mypads%%'
            ORDER BY `key`
            LIMIT %s
            """,
            (args.sample_limit,),
        )

        mypads_blob = "\n".join(
            f"{key}\n{value}" for key, value in mypads_rows
        ).lower()

        pads: list[PadInfo] = []
        for key, value in base_pad_rows:
            pad_id = key[4:]
            head, preview = extract_pad_text(value)
            pads.append(
                PadInfo(
                    pad_id=pad_id,
                    head=head,
                    text_preview=preview,
                    has_mypads_reference=pad_id.lower() in mypads_blob,
                )
            )

        child_type_counts = Counter()
        for key, _ in pad_rows:
            if ":revs:" in key:
                child_type_counts["revisions"] += 1
            elif ":chat:" in key:
                child_type_counts["chat_messages"] += 1
            elif ":readonly:" in key:
                child_type_counts["readonly_mappings"] += 1

        likely_workspace_pads = [p.pad_id for p in pads if p.has_mypads_reference]
        likely_normal_pads = [p.pad_id for p in pads if not p.has_mypads_reference]

        report = {
            "mode": "dry-run",
            "read_only": True,
            "source": {
                "host": args.host,
                "port": args.port,
                "database": database,
                "table": table,
            },
            "summary": {
                "total_store_records": total_records,
                "base_pads_inspected": len(pads),
                "likely_workspace_pads": len(likely_workspace_pads),
                "likely_normal_pads": len(likely_normal_pads),
                "mypads_metadata_samples": len(mypads_rows),
            },
            "prefix_counts": prefix_counts,
            "etherpad_child_record_counts": dict(child_type_counts),
            "classification_warning": (
                "Workspace classification is heuristic. It only checks whether a pad ID "
                "appears in sampled ep_mypads-related records. No data is written."
            ),
            "likely_workspace_pad_ids": likely_workspace_pads,
            "likely_normal_pad_ids": likely_normal_pads,
            "pads": [asdict(p) for p in pads],
            "mypads_metadata_samples": [
                {
                    "key": key,
                    "decoded_json": decode_json(value),
                    "raw_preview": value[:2000],
                }
                for key, value in mypads_rows
            ],
        }

        output = Path(args.output)
        output.write_text(
            json.dumps(report, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )

        connection.rollback()

        print()
        print("DRY-RUN SUMMARY")
        print(f"Store records:             {total_records}")
        print(f"Base pads inspected:       {len(pads)}")
        print(f"Likely workspace pads:     {len(likely_workspace_pads)}")
        print(f"Likely normal pads:        {len(likely_normal_pads)}")
        print(f"MyPads metadata samples:   {len(mypads_rows)}")
        print(f"Report:                    {output.resolve()}")
        print()
        print("No database changes were made.")
        return 0

    except MySQLError as exc:
        LOG.error("MySQL error: %s", exc)
        if connection is not None:
            connection.rollback()
        return 4
    except OSError as exc:
        LOG.error("File error: %s", exc)
        if connection is not None:
            connection.rollback()
        return 5
    finally:
        if connection is not None and connection.is_connected():
            connection.close()


if __name__ == "__main__":
    sys.exit(main())