"""Bounded database executor for verified PChome mapping selectors."""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

from services.pchome_mapping_backlog.artifacts import (
    _canonical_retry_exception_artifact_bytes,
    _resolve_retry_exception_artifact_path,
)
from services.pchome_mapping_backlog.policies import (
    DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY,
)


def _retry_exception_controlled_apply_executor_id(preflight_package: dict[str, Any]) -> str:
    preflight = preflight_package.get("controlled_apply_preflight") or {}
    payload = {
        "preflight_id": preflight.get("preflight_id") or "",
        "run_id": preflight.get("run_id") or "",
        "target_selector_count": (preflight_package.get("summary") or {}).get("target_selector_count") or 0,
        "mutation_plan_count": (preflight_package.get("summary") or {}).get("mutation_plan_count") or 0,
    }
    digest = hashlib.sha256(
        json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
    ).hexdigest()
    return f"pchome-retry-exception-controlled-apply-executor-{digest[:16]}"


def _selector_to_pchome_match_params(selector: dict[str, Any]) -> dict[str, Any]:
    target_pchome_product_id = str(selector.get("target_pchome_product_id") or "").strip()
    return {
        "momo_icode": str(selector.get("momo_product_id") or "").strip(),
        "momo_name": str(selector.get("momo_product_name") or selector.get("momo_product_id") or "").strip(),
        "momo_price": selector.get("momo_price"),
        "pchome_id": target_pchome_product_id,
        "pchome_name": str(selector.get("target_pchome_product_name") or target_pchome_product_id).strip(),
        "pchome_url": f"https://24h.pchome.com.tw/prod/{target_pchome_product_id}" if target_pchome_product_id else None,
        "similarity": selector.get("target_match_score"),
        "advantage": "ai_match",
    }


def _fetch_pchome_product_match_by_momo_icode(conn: Any, momo_icode: str) -> dict[str, Any] | None:
    from sqlalchemy import text

    row = conn.execute(text("""
        SELECT id, momo_name, momo_icode, momo_price, pchome_id, pchome_name,
               pchome_url, pchome_price, pchome_original, pchome_in_stock,
               similarity, price_diff, price_diff_pct, advantage, last_checked
        FROM pchome_product_matches
        WHERE momo_icode = :momo_icode
    """), {"momo_icode": momo_icode}).mappings().first()
    return dict(row) if row else None


def _upsert_pchome_product_match(conn: Any, selector: dict[str, Any]) -> None:
    from sqlalchemy import text

    params = _selector_to_pchome_match_params(selector)
    conn.execute(text("""
        INSERT INTO pchome_product_matches (
            momo_name, momo_icode, momo_price, pchome_id, pchome_name,
            pchome_url, similarity, advantage, last_checked
        )
        VALUES (
            :momo_name, :momo_icode, :momo_price, :pchome_id, :pchome_name,
            :pchome_url, :similarity, :advantage, CURRENT_TIMESTAMP
        )
        ON CONFLICT (momo_icode) DO UPDATE SET
            momo_name = EXCLUDED.momo_name,
            momo_price = EXCLUDED.momo_price,
            pchome_id = EXCLUDED.pchome_id,
            pchome_name = EXCLUDED.pchome_name,
            pchome_url = EXCLUDED.pchome_url,
            similarity = EXCLUDED.similarity,
            advantage = EXCLUDED.advantage,
            last_checked = CURRENT_TIMESTAMP
    """), params)


def build_controlled_apply_executor_from_preflight(
    preflight_package: dict[str, Any],
    *,
    execute_apply: bool = False,
    engine: Any = None,
    materialize_artifacts: bool = False,
    artifact_root: str | Path | None = None,
) -> dict[str, Any]:
    """Apply verified selectors and emit independently verifiable receipts."""
    preflight = preflight_package.get("controlled_apply_preflight") or {}
    selectors = list(preflight_package.get("target_selectors") or [])
    executor_ready = bool(preflight.get("ready_for_controlled_apply_executor")) and bool(selectors)
    executor_id = _retry_exception_controlled_apply_executor_id(preflight_package)
    write_attempted = bool(execute_apply and executor_ready and engine is not None)
    write_blockers: list[str] = []
    if execute_apply and engine is None:
        write_blockers.append("engine_required_for_execute_apply")
    if execute_apply and not executor_ready:
        write_blockers.append("controlled_apply_preflight_not_ready")

    prewrite_snapshots: list[dict[str, Any]] = []
    applied_records: list[dict[str, Any]] = []
    post_apply_readbacks: list[dict[str, Any]] = []
    missing_tables: list[str] = []
    if write_attempted:
        from sqlalchemy import inspect

        with engine.begin() as conn:
            inspector = inspect(conn)
            if not inspector.has_table("pchome_product_matches"):
                missing_tables.append("pchome_product_matches")
            else:
                for selector in selectors:
                    momo_icode = str(selector.get("momo_product_id") or "").strip()
                    before = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode)
                    prewrite_snapshots.append({
                        "selector_id": selector.get("selector_id"),
                        "momo_icode": momo_icode,
                        "before": before,
                        "row_existed_before": before is not None,
                    })
                if not missing_tables:
                    for selector in selectors:
                        _upsert_pchome_product_match(conn, selector)
                        momo_icode = str(selector.get("momo_product_id") or "").strip()
                        after = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode)
                        applied_records.append({
                            "selector_id": selector.get("selector_id"),
                            "momo_icode": momo_icode,
                            "target_pchome_product_id": selector.get("target_pchome_product_id"),
                            "applied": bool(after),
                        })
                        post_apply_readbacks.append({
                            "selector_id": selector.get("selector_id"),
                            "momo_icode": momo_icode,
                            "expected_pchome_id": selector.get("target_pchome_product_id"),
                            "actual_pchome_id": (after or {}).get("pchome_id"),
                            "expected_momo_name": selector.get("momo_product_name"),
                            "actual_momo_name": (after or {}).get("momo_name"),
                            "passed": bool(after)
                            and str((after or {}).get("pchome_id") or "") == str(selector.get("target_pchome_product_id") or "")
                            and str((after or {}).get("momo_icode") or "") == momo_icode,
                            "writes_database": False,
                        })

    readback_pass_count = sum(1 for item in post_apply_readbacks if item.get("passed"))
    applied_count = len(applied_records)
    if execute_apply and missing_tables:
        result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_BLOCKED_MISSING_TABLE"
    elif execute_apply and not selectors:
        result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT"
    elif execute_apply and selectors and applied_count == len(selectors) and readback_pass_count == len(selectors):
        result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED"
    elif execute_apply:
        result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTION_INCOMPLETE"
    elif executor_ready:
        result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_READY"
    else:
        result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT"

    rollback_steps = [
        {
            "selector_id": snapshot.get("selector_id"),
            "momo_icode": snapshot.get("momo_icode"),
            "action": "restore_previous_pchome_product_match" if snapshot.get("row_existed_before") else "delete_inserted_pchome_product_match",
            "prewrite_snapshot_available": True,
            "executes_in_executor": False,
            "writes_database": False,
        }
        for snapshot in prewrite_snapshots
    ] or [
        {
            "selector_id": selector.get("selector_id"),
            "momo_icode": selector.get("momo_product_id"),
            "action": "planned_restore_or_delete_pchome_product_match",
            "prewrite_snapshot_available": False,
            "executes_in_executor": False,
            "writes_database": False,
        }
        for selector in selectors
    ]
    summary = {
        "controlled_apply_executor_ready_count": 1 if executor_ready else 0,
        "execute_apply_requested_count": 1 if execute_apply else 0,
        "target_selector_count": len(selectors),
        "prewrite_snapshot_count": len(prewrite_snapshots),
        "applied_record_count": applied_count,
        "post_apply_readback_count": len(post_apply_readbacks),
        "post_apply_readback_pass_count": readback_pass_count,
        "post_apply_readback_fail_count": len(post_apply_readbacks) - readback_pass_count,
        "rollback_step_count": len(rollback_steps),
        "missing_table_count": len(missing_tables),
        "writes_database_count": applied_count,
        "persists_candidate_count": applied_count,
    }
    executor_metadata = {
        "executor_id": executor_id,
        "source_preflight_id": preflight.get("preflight_id"),
        "run_id": preflight.get("run_id"),
        "stage": "P2_retry_exception_controlled_apply_executor",
        "status": result,
        "execute_apply": bool(execute_apply),
        "target_table": "pchome_product_matches",
        "ready_for_apply": executor_ready,
        "write_attempted": write_attempted,
        "missing_tables": missing_tables,
        "requires_fresh_production_truth": True,
    }
    rollback_plan = {
        "rollback_step_count": len(rollback_steps),
        "rollback_steps": rollback_steps,
        "executes_in_executor": False,
        "writes_database": False,
    }
    safety = {
        "ai_controlled_apply": True,
        "execute_apply": bool(execute_apply),
        "target_table": "pchome_product_matches",
        "writes_database": bool(applied_count),
        "writes_database_count": applied_count,
        "persists_candidate": bool(applied_count),
        "persists_candidate_count": applied_count,
        "syncs_external_offers": False,
        "dispatches_telegram": False,
        "llm_calls_in_executor": False,
        "gemini_allowed": False,
        "requires_production_version_truth": True,
    }
    executor_receipt_ready = (
        result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED"
        and bool(applied_records)
        and readback_pass_count == len(selectors)
    )
    root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data"
    receipt_relative_path = (
        f"artifacts/pchome_growth/retry_exception_closeout/"
        f"controlled_apply_executor/{executor_id}.json"
    )
    executor_receipt_payload = {
        "artifact_key": "retry_exception_controlled_apply_executor_receipt",
        "executor_id": executor_id,
        "source_preflight_id": preflight.get("preflight_id"),
        "run_id": preflight.get("run_id"),
        "source_policy": preflight_package.get("policy"),
        "result": result,
        "created_at": preflight_package.get("generated_at"),
        "summary": summary,
        "controlled_apply_executor": executor_metadata,
        "target_selectors": selectors,
        "prewrite_snapshots": prewrite_snapshots,
        "applied_records": applied_records,
        "post_apply_readbacks": post_apply_readbacks,
        "rollback_plan": rollback_plan,
        "safety": safety,
    }
    executor_receipt_bytes = _canonical_retry_exception_artifact_bytes(executor_receipt_payload)
    executor_receipt_artifact = {
        "key": "retry_exception_controlled_apply_executor_receipt",
        "artifact_type": "controlled_apply_executor_receipt",
        "relative_path": receipt_relative_path,
        "payload_sha256": hashlib.sha256(executor_receipt_bytes).hexdigest(),
        "byte_count": len(executor_receipt_bytes),
        "payload": executor_receipt_payload,
        "materialized": False,
        "writes_database": False,
    }
    materialized_executor_artifacts: list[dict[str, Any]] = []
    if materialize_artifacts and executor_receipt_ready:
        target_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path)
        target_path.parent.mkdir(parents=True, exist_ok=True)
        target_path.write_bytes(executor_receipt_bytes)
        materialized_executor_artifacts.append({
            "key": executor_receipt_artifact["key"],
            "relative_path": receipt_relative_path,
            "absolute_path": str(target_path),
            "payload_sha256": executor_receipt_artifact["payload_sha256"],
            "written_byte_count": target_path.stat().st_size,
            "writes_database": False,
        })
        executor_receipt_artifact["materialized"] = True
        executor_receipt_artifact["absolute_path"] = str(target_path)

    receipt_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path)
    actual_receipt_sha = ""
    receipt_file_exists = receipt_path.exists()
    if receipt_file_exists:
        actual_receipt_sha = hashlib.sha256(receipt_path.read_bytes()).hexdigest()
    receipt_checks = [
        {"check": "executor_receipt_ready_after_apply", "passed": executor_receipt_ready},
        {"check": "all_post_apply_readbacks_passed", "passed": readback_pass_count == len(selectors)},
        {"check": "applied_record_count_matches_selectors", "passed": applied_count == len(selectors)},
        {"check": "receipt_payload_hash_is_sha256", "passed": len(executor_receipt_artifact["payload_sha256"]) == 64},
        {
            "check": "materialized_receipt_exists_when_requested",
            "passed": (not materialize_artifacts) or (executor_receipt_ready and receipt_file_exists),
        },
        {
            "check": "materialized_receipt_hash_matches_expected",
            "passed": (not materialize_artifacts)
            or (bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"]),
        },
        {"check": "receipt_safety_blocks_side_effects", "passed": safety["syncs_external_offers"] is False and safety["dispatches_telegram"] is False},
    ]
    post_executor_receipt_verifier = {
        "ready": executor_receipt_ready,
        "checks": receipt_checks,
        "check_count": len(receipt_checks),
        "passed": all(check.get("passed") is True for check in receipt_checks),
        "expected_sha256": executor_receipt_artifact["payload_sha256"],
        "actual_sha256": actual_receipt_sha,
        "hash_match": bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"],
        "reads_artifact_files": bool(materialize_artifacts),
        "writes_database": False,
    }
    summary["executor_receipt_ready_count"] = 1 if executor_receipt_ready else 0
    summary["executor_receipt_payload_count"] = 1
    summary["executor_receipt_materialized_count"] = len(materialized_executor_artifacts)
    summary["executor_receipt_hash_match_count"] = 1 if post_executor_receipt_verifier["hash_match"] else 0
    summary["post_executor_receipt_verifier_check_count"] = len(receipt_checks)
    safety["writes_artifact_count"] = len(materialized_executor_artifacts)
    return {
        "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY,
        "result": result,
        "success": bool(preflight_package.get("success")),
        "generated_at": preflight_package.get("generated_at"),
        "source_policy": preflight_package.get("policy"),
        "stats": preflight_package.get("stats") or {},
        "backlog": preflight_package.get("backlog") or {},
        "summary": summary,
        "controlled_apply_executor": executor_metadata,
        "target_selectors": selectors,
        "prewrite_snapshots": prewrite_snapshots,
        "applied_records": applied_records,
        "post_apply_readbacks": post_apply_readbacks,
        "rollback_plan": rollback_plan,
        "executor_receipt_artifact": executor_receipt_artifact,
        "materialized_executor_artifacts": materialized_executor_artifacts,
        "post_executor_receipt_verifier": post_executor_receipt_verifier,
        "write_blockers": write_blockers,
        "source_preflight_summary": preflight_package.get("summary") or {},
        "next_actions": [
            "Use the executor receipt artifact as the machine-verifiable closeout source for every applied selector.",
            "Run rollback steps only if post-apply readback fails or a future verifier detects drift.",
            "Keep future writes bounded to selector IDs from this executor package.",
        ],
        "safety": safety,
    }


__all__ = (
    "build_controlled_apply_executor_from_preflight",
)
