"""Controlled PixelRAG marketplace PromotionGate replay worker.

This worker turns marketplace identity matcher receipts into deterministic
PromotionGate replay receipts. It does not fetch marketplace pages, call
models, write databases, write AI insights, or promote candidate prices.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

from services.pixelrag_crawler_integration_service import DEFAULT_ARTIFACT_MAX_AGE_HOURS
from services.pixelrag_marketplace_identity_matcher_replay_service import (
    DEFAULT_OUTPUT_ROOT as DEFAULT_IDENTITY_MATCHER_RECEIPT_ROOT,
    IDENTITY_MATCHER_REPLAY_VERSION,
)


POLICY = "controlled_pixelrag_marketplace_promotion_gate_replay_v1"
PROMOTION_GATE_REPLAY_VERSION = "pixelrag_marketplace_promotion_gate_replay_v1"
DEFAULT_LIMIT = 25
DEFAULT_OUTPUT_ROOT = os.getenv(
    "PIXELRAG_MARKETPLACE_PROMOTION_GATE_REPLAY_RECEIPT_ROOT",
    "/app/data/ai_automation/pixelrag_marketplace_promotion_gate_replay_receipts"
    if Path("/app/data").exists()
    else "runtime_artifacts/pixelrag_marketplace_promotion_gate_replay_receipts",
)

REQUIRED_PROMOTION_GATES = {
    "public_source_boundary",
    "rate_limit_contract",
    "provenance_contract",
    "identity_matcher_replay",
    "promotion_gate",
    "embedding_signature_guard",
}


def _normalise_platforms(
    platform: str | tuple[str, ...] | list[str] | None,
) -> tuple[str, ...]:
    if isinstance(platform, str):
        value = platform.strip().lower()
        return (value,) if value else ()
    return tuple(
        str(item or "").strip().lower()
        for item in (platform or ())
        if str(item or "").strip()
    )


def _safe_segment(value: Any) -> str:
    text = str(value or "unknown").strip().lower()
    text = re.sub(r"[^a-z0-9._-]+", "-", text)
    return text.strip("-") or "unknown"


def _parse_iso_datetime(value: Any) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None


def _receipt_candidates(
    root: Path,
    *,
    platforms: tuple[str, ...],
    limit: int,
) -> list[Path]:
    if not root.exists():
        return []
    candidates: list[Path] = []
    if platforms:
        for platform in platforms:
            candidates.extend(
                (root / platform).glob("*/marketplace_identity_matcher_replay_receipt.json")
            )
    else:
        candidates.extend(root.glob("*/*/marketplace_identity_matcher_replay_receipt.json"))
    return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit]


def _load_receipt(path: Path) -> tuple[dict[str, Any], list[str]]:
    try:
        return json.loads(path.read_text(encoding="utf-8")), []
    except (OSError, json.JSONDecodeError) as exc:
        return {}, [str(exc)[:300]]


def _as_mapping(value: Any) -> Mapping[str, Any]:
    return value if isinstance(value, Mapping) else {}


def _as_list(value: Any) -> list[Any]:
    return list(value) if isinstance(value, list) else []


def _required_before(identity_payload: Mapping[str, Any]) -> set[str]:
    return set(str(item) for item in _as_list(identity_payload.get("required_before_data_promotion")))


def _promotion_fingerprint(
    platform: str,
    manifest_id: str,
    candidate_id: str,
    identity_fingerprint: str,
) -> str:
    digest = hashlib.sha256(
        f"{platform}:{manifest_id}:{candidate_id}:{identity_fingerprint}".encode("utf-8")
    ).hexdigest()
    return digest[:24]


def _promotion_candidate_contracts(
    receipt: Mapping[str, Any],
    *,
    identity_payload: Mapping[str, Any],
) -> list[dict[str, Any]]:
    platform = str(receipt.get("platform") or "unknown")
    manifest_id = str(receipt.get("manifest_id") or "unknown")
    contracts: list[dict[str, Any]] = []
    for candidate in _as_list(identity_payload.get("identity_candidate_contracts")):
        if not isinstance(candidate, Mapping):
            continue
        candidate_id = str(candidate.get("candidate_id") or f"{platform}:{manifest_id}:unknown")
        identity_fingerprint = str(candidate.get("identity_fingerprint") or "missing")
        contracts.append(
            {
                "candidate_id": candidate_id,
                "candidate_index": int(candidate.get("candidate_index") or 0),
                "identity_fingerprint": identity_fingerprint,
                "promotion_fingerprint": _promotion_fingerprint(
                    platform,
                    manifest_id,
                    candidate_id,
                    identity_fingerprint,
                ),
                "promotion_strategy": "deterministic_gate_replay_no_values",
                "promotion_stage": "pre_embedding_signature_guard",
                "requires_public_source_boundary": True,
                "requires_rate_limit_contract": True,
                "requires_source_provenance": True,
                "requires_identity_matcher_replay": True,
                "requires_promotion_gate": True,
                "requires_embedding_signature_guard": True,
                "requires_evidence_time": True,
                "ready_for_embedding_signature_guard_replay": True,
                "ready_for_candidate_knowledge_replay": True,
                "ready_for_price_table_write": False,
                "price_write_blocked_until_embedding_signature_guard": True,
            }
        )
    return contracts


def _promotion_gate_payload(
    receipt: Mapping[str, Any],
    *,
    receipt_path: Path,
) -> dict[str, Any]:
    identity_payload = _as_mapping(receipt.get("identity_matcher_replay"))
    promotion_contracts = _promotion_candidate_contracts(
        receipt,
        identity_payload=identity_payload,
    )
    return {
        "promotion_gate_replay_version": PROMOTION_GATE_REPLAY_VERSION,
        "identity_matcher_replay_version": identity_payload.get(
            "identity_matcher_replay_version"
        ),
        "adapter_dry_run_version": identity_payload.get("adapter_dry_run_version"),
        "adapter_preflight_version": identity_payload.get("adapter_preflight_version"),
        "source_contract_version": identity_payload.get("source_contract_version"),
        "adapter_code": identity_payload.get("adapter_code") or receipt.get("adapter_code"),
        "contract_id": identity_payload.get("contract_id"),
        "barrier_type": identity_payload.get("barrier_type"),
        "capture_runtime_unavailable": bool(
            identity_payload.get("capture_runtime_unavailable")
        ),
        "input_identity_matcher_receipt_path": str(receipt_path),
        "input_adapter_dry_run_receipt_path": identity_payload.get(
            "input_adapter_dry_run_receipt_path"
        ),
        "adapter_preflight_receipt_path": identity_payload.get(
            "adapter_preflight_receipt_path"
        ),
        "source_contract_replay_receipt_path": identity_payload.get(
            "source_contract_replay_receipt_path"
        ),
        "source_worker_receipt_path": identity_payload.get("source_worker_receipt_path"),
        "source_receipt_path": identity_payload.get("source_receipt_path"),
        "promotion_gate_execution_mode": "deterministic_replay_only_no_network_no_db",
        "promotion_candidate_count": len(promotion_contracts),
        "promotion_candidate_contracts": promotion_contracts,
        "required_before_data_promotion": list(
            identity_payload.get("required_before_data_promotion") or []
        ),
        "allowed_next_step": "run_marketplace_embedding_signature_guard_replay",
    }


def _promotion_gate_checks(
    receipt: Mapping[str, Any],
    *,
    stale: bool,
    errors: list[str],
    promotion_payload: Mapping[str, Any],
) -> dict[str, bool]:
    identity_payload = _as_mapping(receipt.get("identity_matcher_replay"))
    boundary = _as_mapping(receipt.get("promotion_boundary"))
    identity_checks = _as_mapping(receipt.get("identity_matcher_replay_checks"))
    identity_contracts = _as_list(identity_payload.get("identity_candidate_contracts"))
    promotion_contracts = _as_list(promotion_payload.get("promotion_candidate_contracts"))
    required_before = _required_before(identity_payload)
    return {
        "receipt_parse_ok": not errors,
        "receipt_fresh": not stale,
        "identity_matcher_replay_ready": (
            receipt.get("worker_status")
            == "executed_marketplace_identity_matcher_replay_ready"
        ),
        "identity_matcher_version_supported": (
            identity_payload.get("identity_matcher_replay_version")
            == IDENTITY_MATCHER_REPLAY_VERSION
        ),
        "identity_matcher_checks_all_passed": (
            int(receipt.get("identity_matcher_replay_check_pass_count") or 0)
            == int(receipt.get("identity_matcher_replay_check_count") or -1)
            and int(receipt.get("identity_matcher_replay_check_count") or 0) > 0
        ),
        "blocked_page_not_product_data": bool(
            identity_checks.get("blocked_page_not_product_data")
        ),
        "identity_candidate_contracts_present": bool(identity_contracts),
        "promotion_candidate_contracts_generated": bool(promotion_contracts),
        "promotion_candidate_count_matches_identity_count": (
            len(promotion_contracts) == len(identity_contracts)
        ),
        "identity_candidates_ready_for_promotion": all(
            bool(item.get("ready_for_promotion_gate_replay"))
            for item in identity_contracts
        ),
        "required_promotion_gates_present": REQUIRED_PROMOTION_GATES.issubset(
            required_before
        ),
        "promotion_gate_required": "promotion_gate" in required_before,
        "embedding_signature_guard_required": "embedding_signature_guard" in required_before,
        "embedding_signature_guard_replay_ready": all(
            bool(item.get("ready_for_embedding_signature_guard_replay"))
            for item in promotion_contracts
        ),
        "candidate_knowledge_replay_ready": all(
            bool(item.get("ready_for_candidate_knowledge_replay"))
            for item in promotion_contracts
        ),
        "price_write_blocked_until_embedding_signature_guard": all(
            bool(item.get("price_write_blocked_until_embedding_signature_guard"))
            and not bool(item.get("ready_for_price_table_write"))
            for item in promotion_contracts
        ),
        "direct_price_write_disallowed": not bool(
            boundary.get("direct_price_write_allowed")
        ),
        "direct_ai_insights_write_disallowed": not bool(
            boundary.get("direct_ai_insights_write_allowed")
        ),
        "database_write_disallowed": not bool(receipt.get("writes_database"))
        and int(receipt.get("writes_database_count") or 0) == 0,
        "network_call_disallowed": not bool(receipt.get("network_call_performed")),
        "model_call_disallowed": not bool(receipt.get("model_call_performed")),
        "source_provenance_paths_present": bool(
            promotion_payload.get("input_identity_matcher_receipt_path")
            and promotion_payload.get("input_adapter_dry_run_receipt_path")
            and promotion_payload.get("adapter_preflight_receipt_path")
            and promotion_payload.get("source_contract_replay_receipt_path")
        ),
    }


def _worker_item(
    receipt_path: Path,
    *,
    now: datetime,
    max_age_hours: int,
    execute: bool,
) -> dict[str, Any]:
    receipt, errors = _load_receipt(receipt_path)
    generated_at = _parse_iso_datetime(receipt.get("generated_at"))
    if generated_at is None:
        try:
            generated_at = datetime.fromtimestamp(receipt_path.stat().st_mtime, tz=timezone.utc)
        except OSError:
            generated_at = None
    age_hours = ((now - generated_at).total_seconds() / 3600) if generated_at else None
    stale = age_hours is None or age_hours > max_age_hours
    promotion_payload = _promotion_gate_payload(receipt, receipt_path=receipt_path)
    checks = _promotion_gate_checks(
        receipt,
        stale=stale,
        errors=errors,
        promotion_payload=promotion_payload,
    )
    check_count = len(checks)
    pass_count = sum(1 for passed in checks.values() if passed)
    ready = pass_count == check_count
    platform = str(receipt.get("platform") or receipt_path.parent.parent.name).strip().lower()
    manifest_id = str(receipt.get("manifest_id") or receipt_path.parent.name).strip()
    status = (
        "executed_marketplace_promotion_gate_replay_ready"
        if execute and ready
        else (
            "dry_run_ready_for_marketplace_promotion_gate_replay"
            if ready
            else "skipped_marketplace_promotion_gate_replay_guard_failed"
        )
    )
    return {
        "worker_status": status,
        "platform": platform,
        "manifest_id": manifest_id,
        "source_type": "marketplace_identity_matcher_replay_receipt",
        "identity_matcher_receipt_path": str(receipt_path),
        "adapter_dry_run_receipt_path": promotion_payload.get(
            "input_adapter_dry_run_receipt_path"
        ),
        "adapter_preflight_receipt_path": promotion_payload.get(
            "adapter_preflight_receipt_path"
        ),
        "source_contract_replay_receipt_path": promotion_payload.get(
            "source_contract_replay_receipt_path"
        ),
        "source_worker_receipt_path": promotion_payload.get("source_worker_receipt_path"),
        "source_receipt_path": promotion_payload.get("source_receipt_path"),
        "adapter_code": promotion_payload.get("adapter_code"),
        "promotion_gate_replay_status": "ready" if ready else "blocked",
        "ready_for_execution": ready,
        "execute": bool(execute),
        "generated_at": generated_at.isoformat() if generated_at else None,
        "age_hours": round(age_hours, 3) if age_hours is not None else None,
        "stale": stale,
        "network_call_performed": False,
        "model_call_performed": False,
        "artifact_write_performed": False,
        "writes_database": False,
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
        "promotion_gate_replay_checks": checks,
        "promotion_gate_replay_check_count": check_count,
        "promotion_gate_replay_check_pass_count": pass_count,
        "promotion_gate_replay": promotion_payload,
        "promotion_boundary": {
            "direct_ai_insights_write_allowed": False,
            "direct_price_write_allowed": False,
            "promotion_gate_replay_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_embedding_signature_guard": True,
        },
        "next_machine_action": (
            "run_marketplace_embedding_signature_guard_replay"
            if execute and ready
            else (
                "run_marketplace_promotion_gate_replay_execute"
                if ready
                else "repair_marketplace_promotion_gate_replay_inputs"
            )
        ),
    }


def _write_promotion_gate_receipt(*, output_root: Path, item: Mapping[str, Any]) -> str:
    target = (
        output_root
        / _safe_segment(item.get("platform"))
        / _safe_segment(item.get("manifest_id"))
        / "marketplace_promotion_gate_replay_receipt.json"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    receipt = dict(item)
    receipt["artifact_write_performed"] = True
    receipt["receipt_path"] = str(target)
    receipt["generated_at"] = datetime.now(timezone.utc).isoformat()
    receipt["policy"] = POLICY
    target.write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    return str(target)


def run_pixelrag_marketplace_promotion_gate_replay(
    *,
    identity_matcher_receipt_root: str | Path | None = None,
    output_root: str | Path | None = None,
    platform: str | tuple[str, ...] | list[str] | None = None,
    max_age_hours: int | None = None,
    limit: int | None = None,
    execute: bool = False,
    write_receipt: bool = False,
) -> dict[str, Any]:
    """Run or dry-run marketplace PromotionGate replay."""
    source_root = Path(identity_matcher_receipt_root or DEFAULT_IDENTITY_MATCHER_RECEIPT_ROOT)
    output = Path(output_root or DEFAULT_OUTPUT_ROOT)
    platforms = _normalise_platforms(platform)
    max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS))
    item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 250))
    now = datetime.now(timezone.utc)
    receipt_paths = _receipt_candidates(source_root, platforms=platforms, limit=item_limit)
    worker_items: list[dict[str, Any]] = []
    for receipt_path in receipt_paths:
        item = _worker_item(
            receipt_path,
            now=now,
            max_age_hours=max_age,
            execute=execute,
        )
        if execute and write_receipt and item.get("ready_for_execution"):
            item["receipt_path"] = _write_promotion_gate_receipt(
                output_root=output,
                item=item,
            )
            item["artifact_write_performed"] = True
        worker_items.append(item)

    ready_count = sum(1 for item in worker_items if item.get("ready_for_execution"))
    blocked_count = len(worker_items) - ready_count
    dry_run_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("dry_run_")
    )
    executed_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("executed_")
    )
    receipt_written_count = sum(1 for item in worker_items if item.get("receipt_path"))
    guard_failed_count = sum(
        1 for item in worker_items if item.get("promotion_gate_replay_status") != "ready"
    )
    capture_runtime_gap_count = sum(
        1
        for item in worker_items
        if (item.get("promotion_gate_replay") or {}).get("capture_runtime_unavailable")
    )
    promotion_candidate_count = sum(
        int((item.get("promotion_gate_replay") or {}).get("promotion_candidate_count") or 0)
        for item in worker_items
    )
    embedding_signature_guard_ready_count = sum(
        sum(
            1
            for candidate in _as_list(
                (item.get("promotion_gate_replay") or {}).get(
                    "promotion_candidate_contracts"
                )
            )
            if isinstance(candidate, Mapping)
            and candidate.get("ready_for_embedding_signature_guard_replay")
        )
        for item in worker_items
    )

    if not worker_items:
        status = "warning"
    elif guard_failed_count:
        status = "warning"
    else:
        status = "ok"

    if not worker_items:
        next_action = "run_marketplace_adapter_identity_matcher_replay"
    elif not execute and ready_count:
        next_action = "run_marketplace_promotion_gate_replay_execute"
    elif execute and receipt_written_count:
        next_action = "run_marketplace_embedding_signature_guard_replay"
    elif guard_failed_count:
        next_action = "repair_marketplace_promotion_gate_replay_inputs"
    else:
        next_action = "refresh_marketplace_promotion_gate_replay_candidates"

    summary = {
        "candidate_count": len(worker_items),
        "ready_count": ready_count,
        "blocked_count": blocked_count,
        "dry_run_count": dry_run_count,
        "executed_count": executed_count,
        "receipt_written_count": receipt_written_count,
        "guard_failed_count": guard_failed_count,
        "capture_runtime_gap_count": capture_runtime_gap_count,
        "promotion_candidate_count": promotion_candidate_count,
        "embedding_signature_guard_ready_count": embedding_signature_guard_ready_count,
        "platforms": sorted({str(item.get("platform") or "unknown") for item in worker_items}),
        "network_call_performed": False,
        "model_call_performed": False,
        "artifact_write_performed": bool(receipt_written_count),
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
    }
    return {
        "success": status != "critical",
        "policy": POLICY,
        "status": status,
        "generated_at": now.isoformat(),
        "promotion_gate_replay_version": PROMOTION_GATE_REPLAY_VERSION,
        "identity_matcher_receipt_root": str(source_root),
        "output_root": str(output),
        "platform_filter": list(platforms),
        "max_age_hours": max_age,
        "limit": item_limit,
        "execute": bool(execute),
        "write_receipt": bool(write_receipt and execute),
        "summary": summary,
        "worker_items": worker_items,
        "controlled_apply": {
            "network_call": False,
            "model_call": False,
            "artifact_write": bool(receipt_written_count),
            "db_write": False,
            "writes_database": False,
            "writes_database_count": 0,
            "secret_read": False,
            "raw_cookie_or_session_read": False,
            "credentialed_session_allowed": False,
            "login_allowed": False,
            "production_price_write": False,
            "primary_human_gate_count": 0,
        },
        "promotion_boundary": {
            "writes_ai_insights": False,
            "writes_price_tables": False,
            "promotion_gate_replay_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_embedding_signature_guard": True,
        },
        "next_machine_action": next_action,
    }


__all__ = [
    "DEFAULT_OUTPUT_ROOT",
    "POLICY",
    "PROMOTION_GATE_REPLAY_VERSION",
    "run_pixelrag_marketplace_promotion_gate_replay",
]
