"""Z01 기초단가 저장소 — `master_base_price` · `master_revision` (020 · PLAN 1-0 계약 · 2026-09-18).

정본 = DB. 줄 = (kind, row_key) 한 줄 · `data` 는 지금 값 한 줄 전체 · `seed` 는 주입 때 초기값(관리자 추가 줄은 NULL).
판번호 = `master_revision.revision`(scope `base:{kind}`) — 쓸 때마다 +1 · 읽어 간 판과 다르면 안 씀(409).
⚠ 여기는 SQL 만 — 무엇을 고칠 수 있나·값 검사는 `Z01_MasterData_BasePrices` 가 함.
"""

from __future__ import annotations

import json
from typing import Any

import aiomysql

from config.config_db import get_db_pool


class RevisionConflict(Exception):
    """읽어 간 판번호가 지금 판과 다름 — 그 사이 누가 저장함."""


class UnknownRows(LookupError):
    """없는 줄 열쇠 — 되돌릴 줄을 못 찾음."""


def _scope(kind: str) -> str:
    return f"base:{kind}"


def _json(text: Any) -> Any:
    return json.loads(text) if isinstance(text, (str, bytes)) else text


def _dump(data: dict[str, Any]) -> str:
    return json.dumps(data, ensure_ascii=False, default=str)


async def load(kind: str) -> list[dict[str, Any]]:
    """지운 줄 뺀 전부 — 넣은 차례(id) 그대로. {row_key, data, seed}."""
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """SELECT row_key, data, seed FROM master_base_price
               WHERE kind = %s AND is_deleted = 0 ORDER BY id""",
            (kind,),
        )
        return [
            {"row_key": r["row_key"], "data": _json(r["data"]), "seed": _json(r["seed"])}
            for r in await cursor.fetchall()
        ]


async def revision(kind: str) -> int:
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            "SELECT revision FROM master_revision WHERE scope = %s", (_scope(kind),)
        )
        row = await cursor.fetchone()
        return int(row[0]) if row else 0


async def counts() -> dict[str, int]:
    """kind 마다 줄 수(지운 줄 뺌) — 표 목록 한 번에."""
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            "SELECT kind, COUNT(*) FROM master_base_price WHERE is_deleted = 0 GROUP BY kind"
        )
        return {kind: int(n) for kind, n in await cursor.fetchall()}


async def _lock_revision(cursor: Any, kind: str) -> int:
    """판번호 줄을 잠그고 읽음 — 없으면 0 으로 세움(같은 kind 쓰기는 여기서 한 줄로 섬)."""
    await cursor.execute(
        "INSERT IGNORE INTO master_revision (scope, revision) VALUES (%s, 0)", (_scope(kind),)
    )
    await cursor.execute(
        "SELECT revision FROM master_revision WHERE scope = %s FOR UPDATE", (_scope(kind),)
    )
    return int((await cursor.fetchone())[0])


async def _bump(cursor: Any, kind: str) -> None:
    await cursor.execute(
        "UPDATE master_revision SET revision = revision + 1 WHERE scope = %s", (_scope(kind),)
    )


async def save(
    kind: str,
    base_revision: int,
    changed: dict[str, dict[str, Any]],
    added: dict[str, dict[str, Any]],
    deleted: list[str],
    by: int | None,
) -> int:
    """한 트랜잭션 — 판이 맞을 때만 고친 줄·새 줄·지운 줄을 한꺼번에(전부 아니면 전무). 새 판번호."""
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        try:
            await connection.begin()
            current = await _lock_revision(cursor, kind)
            if current != base_revision:
                raise RevisionConflict(current)
            for row_key, data in changed.items():
                await cursor.execute(
                    """UPDATE master_base_price
                       SET data = %s, updated_by = %s, updated_at = CURRENT_TIMESTAMP
                       WHERE kind = %s AND row_key = %s AND is_deleted = 0""",
                    (_dump(data), by, kind, row_key),
                )
            for row_key, data in added.items():
                await cursor.execute(
                    """INSERT INTO master_base_price
                         (kind, row_key, data, seed, edition, is_deleted, updated_by, updated_at)
                       VALUES (%s, %s, %s, NULL, %s, 0, %s, CURRENT_TIMESTAMP)""",
                    (kind, row_key, _dump(data), str(data.get("effective_date") or ""), by),
                )
            for row_key in deleted:
                await cursor.execute(
                    """UPDATE master_base_price
                       SET is_deleted = 1, updated_by = %s, updated_at = CURRENT_TIMESTAMP
                       WHERE kind = %s AND row_key = %s""",
                    (by, kind, row_key),
                )
            await _bump(cursor, kind)
            await connection.commit()
            return current + 1
        except Exception:
            await connection.rollback()
            raise


async def reset(kind: str, row_keys: list[str], by: int | None) -> int:
    """고른 줄을 초기값으로 — 주입 줄은 data := seed(지운 줄도 되살림) · 추가 줄(seed NULL)은 지움. 새 판번호."""
    keys = list(dict.fromkeys(row_keys))
    marks = ", ".join(["%s"] * len(keys))
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        try:
            await connection.begin()
            current = await _lock_revision(cursor, kind)
            await cursor.execute(
                f"SELECT row_key FROM master_base_price WHERE kind = %s AND row_key IN ({marks})",
                (kind, *keys),
            )
            missing = set(keys) - {r[0] for r in await cursor.fetchall()}
            if missing:
                raise UnknownRows(sorted(missing))
            await cursor.execute(
                f"""UPDATE master_base_price
                    SET data = seed, is_deleted = 0, updated_by = %s, updated_at = CURRENT_TIMESTAMP
                    WHERE kind = %s AND seed IS NOT NULL AND row_key IN ({marks})""",
                (by, kind, *keys),
            )
            await cursor.execute(
                f"""UPDATE master_base_price
                    SET is_deleted = 1, updated_by = %s, updated_at = CURRENT_TIMESTAMP
                    WHERE kind = %s AND seed IS NULL AND row_key IN ({marks})""",
                (by, kind, *keys),
            )
            await _bump(cursor, kind)
            await connection.commit()
            return current + 1
        except Exception:
            await connection.rollback()
            raise