"""Controlled PixelRAG marketplace embedding signature guard replay worker.

This worker turns marketplace PromotionGate receipts into deterministic
embedding signature guard receipts. It does not generate embeddings, 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_promotion_gate_replay_service import (
    DEFAULT_OUTPUT_ROOT as DEFAULT_PROMOTION_GATE_RECEIPT_ROOT,
    PROMOTION_GATE_REPLAY_VERSION,
)
from services.rag_service import (
    RAG_EMBED_DIM,
    RAG_EMBED_MODEL,
    RAG_EMBED_NORMALIZE,
    get_embedding_signature,
)


POLICY = "controlled_pixelrag_marketplace_embedding_signature_guard_replay_v1"
EMBEDDING_SIGNATURE_GUARD_REPLAY_VERSION = (
    "pixelrag_marketplace_embedding_signature_guard_replay_v1"
)
DEFAULT_LIMIT = 25
DEFAULT_OUTPUT_ROOT = os.getenv(
    "PIXELRAG_MARKETPLACE_EMBEDDING_SIGNATURE_GUARD_REPLAY_RECEIPT_ROOT",
    "/app/data/ai_automation/pixelrag_marketplace_embedding_signature_guard_replay_receipts"
    if Path("/app/data").exists()
    else "runtime_artifacts/pixelrag_marketplace_embedding_signature_guard_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_promotion_gate_replay_receipt.json")
            )
    else:
        candidates.extend(root.glob("*/*/marketplace_promotion_gate_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(promotion_payload: Mapping[str, Any]) -> set[str]:
    return set(str(item) for item in _as_list(promotion_payload.get("required_before_data_promotion")))


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


def _embedding_signature_contract() -> dict[str, Any]:
    signature = get_embedding_signature()
    return {
        "embedding_signature": signature,
        "embedding_model": RAG_EMBED_MODEL,
        "embedding_dim": RAG_EMBED_DIM,
        "embedding_normalize": RAG_EMBED_NORMALIZE,
        "signature_algorithm": "sha1(model|normalize|dim)[:12]",
        "embedding_generation_performed": False,
        "model_call_performed": False,
    }


def _signature_guard_candidate_contracts(
    receipt: Mapping[str, Any],
    *,
    promotion_payload: Mapping[str, Any],
    embedding_contract: Mapping[str, Any],
) -> list[dict[str, Any]]:
    platform = str(receipt.get("platform") or "unknown")
    manifest_id = str(receipt.get("manifest_id") or "unknown")
    embedding_signature = str(embedding_contract.get("embedding_signature") or "")
    contracts: list[dict[str, Any]] = []
    for candidate in _as_list(promotion_payload.get("promotion_candidate_contracts")):
        if not isinstance(candidate, Mapping):
            continue
        candidate_id = str(candidate.get("candidate_id") or f"{platform}:{manifest_id}:unknown")
        promotion_fingerprint = str(candidate.get("promotion_fingerprint") or "missing")
        contracts.append(
            {
                "candidate_id": candidate_id,
                "candidate_index": int(candidate.get("candidate_index") or 0),
                "promotion_fingerprint": promotion_fingerprint,
                "identity_fingerprint": candidate.get("identity_fingerprint"),
                "expected_embedding_signature": embedding_signature,
                "embedding_signature": embedding_signature,
                "embedding_signature_guard_fingerprint": _signature_guard_fingerprint(
                    platform,
                    manifest_id,
                    candidate_id,
                    promotion_fingerprint,
                    embedding_signature,
                ),
                "signature_guard_strategy": "deterministic_signature_contract_no_embedding_generation",
                "guard_stage": "pre_candidate_knowledge_replay",
                "requires_embedding_signature_guard": True,
                "requires_candidate_knowledge_replay": True,
                "requires_source_provenance": True,
                "requires_replay_canary_before_db_write": True,
                "embedding_generation_performed": False,
                "ready_for_candidate_knowledge_replay": True,
                "ready_for_ai_insights_write": False,
                "ready_for_price_table_write": False,
                "price_write_blocked_until_candidate_canary": True,
            }
        )
    return contracts


def _embedding_signature_guard_payload(
    receipt: Mapping[str, Any],
    *,
    receipt_path: Path,
) -> dict[str, Any]:
    promotion_payload = _as_mapping(receipt.get("promotion_gate_replay"))
    embedding_contract = _embedding_signature_contract()
    guard_contracts = _signature_guard_candidate_contracts(
        receipt,
        promotion_payload=promotion_payload,
        embedding_contract=embedding_contract,
    )
    return {
        "embedding_signature_guard_replay_version": (
            EMBEDDING_SIGNATURE_GUARD_REPLAY_VERSION
        ),
        "promotion_gate_replay_version": promotion_payload.get(
            "promotion_gate_replay_version"
        ),
        "identity_matcher_replay_version": promotion_payload.get(
            "identity_matcher_replay_version"
        ),
        "adapter_dry_run_version": promotion_payload.get("adapter_dry_run_version"),
        "adapter_preflight_version": promotion_payload.get("adapter_preflight_version"),
        "source_contract_version": promotion_payload.get("source_contract_version"),
        "adapter_code": promotion_payload.get("adapter_code") or receipt.get("adapter_code"),
        "contract_id": promotion_payload.get("contract_id"),
        "barrier_type": promotion_payload.get("barrier_type"),
        "capture_runtime_unavailable": bool(
            promotion_payload.get("capture_runtime_unavailable")
        ),
        "input_promotion_gate_receipt_path": str(receipt_path),
        "input_identity_matcher_receipt_path": promotion_payload.get(
            "input_identity_matcher_receipt_path"
        ),
        "input_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"),
        "embedding_signature_contract": embedding_contract,
        "embedding_signature_guard_execution_mode": (
            "deterministic_signature_guard_only_no_embedding_no_network_no_db"
        ),
        "signature_guard_candidate_count": len(guard_contracts),
        "signature_guard_candidate_contracts": guard_contracts,
        "required_before_data_promotion": list(
            promotion_payload.get("required_before_data_promotion") or []
        ),
        "allowed_next_step": "run_marketplace_candidate_knowledge_replay",
    }


def _embedding_signature_guard_checks(
    receipt: Mapping[str, Any],
    *,
    stale: bool,
    errors: list[str],
    guard_payload: Mapping[str, Any],
) -> dict[str, bool]:
    promotion_payload = _as_mapping(receipt.get("promotion_gate_replay"))
    boundary = _as_mapping(receipt.get("promotion_boundary"))
    promotion_checks = _as_mapping(receipt.get("promotion_gate_replay_checks"))
    promotion_contracts = _as_list(promotion_payload.get("promotion_candidate_contracts"))
    guard_contracts = _as_list(guard_payload.get("signature_guard_candidate_contracts"))
    embedding_contract = _as_mapping(guard_payload.get("embedding_signature_contract"))
    expected_signature = str(embedding_contract.get("embedding_signature") or "")
    required_before = _required_before(promotion_payload)
    return {
        "receipt_parse_ok": not errors,
        "receipt_fresh": not stale,
        "promotion_gate_replay_ready": (
            receipt.get("worker_status")
            == "executed_marketplace_promotion_gate_replay_ready"
        ),
        "promotion_gate_version_supported": (
            promotion_payload.get("promotion_gate_replay_version")
            == PROMOTION_GATE_REPLAY_VERSION
        ),
        "promotion_gate_checks_all_passed": (
            int(receipt.get("promotion_gate_replay_check_pass_count") or 0)
            == int(receipt.get("promotion_gate_replay_check_count") or -1)
            and int(receipt.get("promotion_gate_replay_check_count") or 0) > 0
        ),
        "blocked_page_not_product_data": bool(
            promotion_checks.get("blocked_page_not_product_data")
        ),
        "promotion_candidate_contracts_present": bool(promotion_contracts),
        "signature_guard_contracts_generated": bool(guard_contracts),
        "signature_guard_count_matches_promotion_count": (
            len(guard_contracts) == len(promotion_contracts)
        ),
        "promotion_candidates_ready_for_embedding_signature_guard": all(
            bool(item.get("ready_for_embedding_signature_guard_replay"))
            for item in promotion_contracts
        ),
        "required_promotion_gates_present": REQUIRED_PROMOTION_GATES.issubset(
            required_before
        ),
        "embedding_signature_guard_required": "embedding_signature_guard" in required_before,
        "embedding_signature_present": (
            len(expected_signature) == 12
            and all(ch in "0123456789abcdef" for ch in expected_signature)
        ),
        "embedding_generation_not_performed": (
            not bool(embedding_contract.get("embedding_generation_performed"))
            and not bool(embedding_contract.get("model_call_performed"))
        ),
        "signature_guard_all_match_expected_signature": all(
            item.get("embedding_signature") == expected_signature
            and item.get("expected_embedding_signature") == expected_signature
            for item in guard_contracts
        ),
        "candidate_knowledge_replay_ready": all(
            bool(item.get("ready_for_candidate_knowledge_replay"))
            for item in guard_contracts
        ),
        "ai_insights_write_disallowed": all(
            not bool(item.get("ready_for_ai_insights_write"))
            for item in guard_contracts
        ),
        "price_write_blocked_until_candidate_canary": all(
            bool(item.get("price_write_blocked_until_candidate_canary"))
            and not bool(item.get("ready_for_price_table_write"))
            for item in guard_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(
            guard_payload.get("input_promotion_gate_receipt_path")
            and guard_payload.get("input_identity_matcher_receipt_path")
            and guard_payload.get("input_adapter_dry_run_receipt_path")
            and guard_payload.get("adapter_preflight_receipt_path")
            and guard_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
    guard_payload = _embedding_signature_guard_payload(receipt, receipt_path=receipt_path)
    checks = _embedding_signature_guard_checks(
        receipt,
        stale=stale,
        errors=errors,
        guard_payload=guard_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_embedding_signature_guard_replay_ready"
        if execute and ready
        else (
            "dry_run_ready_for_marketplace_embedding_signature_guard_replay"
            if ready
            else "skipped_marketplace_embedding_signature_guard_replay_guard_failed"
        )
    )
    return {
        "worker_status": status,
        "platform": platform,
        "manifest_id": manifest_id,
        "source_type": "marketplace_promotion_gate_replay_receipt",
        "promotion_gate_receipt_path": str(receipt_path),
        "identity_matcher_receipt_path": guard_payload.get(
            "input_identity_matcher_receipt_path"
        ),
        "adapter_dry_run_receipt_path": guard_payload.get(
            "input_adapter_dry_run_receipt_path"
        ),
        "adapter_preflight_receipt_path": guard_payload.get(
            "adapter_preflight_receipt_path"
        ),
        "source_contract_replay_receipt_path": guard_payload.get(
            "source_contract_replay_receipt_path"
        ),
        "source_worker_receipt_path": guard_payload.get("source_worker_receipt_path"),
        "source_receipt_path": guard_payload.get("source_receipt_path"),
        "adapter_code": guard_payload.get("adapter_code"),
        "embedding_signature_guard_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,
        "embedding_generation_performed": False,
        "artifact_write_performed": False,
        "writes_database": False,
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
        "embedding_signature_guard_replay_checks": checks,
        "embedding_signature_guard_replay_check_count": check_count,
        "embedding_signature_guard_replay_check_pass_count": pass_count,
        "embedding_signature_guard_replay": guard_payload,
        "promotion_boundary": {
            "direct_ai_insights_write_allowed": False,
            "direct_price_write_allowed": False,
            "embedding_signature_guard_replay_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_candidate_knowledge_replay": True,
        },
        "next_machine_action": (
            "run_marketplace_candidate_knowledge_replay"
            if execute and ready
            else (
                "run_marketplace_embedding_signature_guard_replay_execute"
                if ready
                else "repair_marketplace_embedding_signature_guard_replay_inputs"
            )
        ),
    }


def _write_embedding_signature_guard_receipt(
    *,
    output_root: Path,
    item: Mapping[str, Any],
) -> str:
    target = (
        output_root
        / _safe_segment(item.get("platform"))
        / _safe_segment(item.get("manifest_id"))
        / "marketplace_embedding_signature_guard_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_embedding_signature_guard_replay(
    *,
    promotion_gate_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 embedding signature guard replay."""
    source_root = Path(promotion_gate_receipt_root or DEFAULT_PROMOTION_GATE_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_embedding_signature_guard_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("embedding_signature_guard_replay_status") != "ready"
    )
    capture_runtime_gap_count = sum(
        1
        for item in worker_items
        if (item.get("embedding_signature_guard_replay") or {}).get(
            "capture_runtime_unavailable"
        )
    )
    signature_guard_candidate_count = sum(
        int(
            (item.get("embedding_signature_guard_replay") or {}).get(
                "signature_guard_candidate_count"
            )
            or 0
        )
        for item in worker_items
    )
    candidate_knowledge_ready_count = sum(
        sum(
            1
            for candidate in _as_list(
                (item.get("embedding_signature_guard_replay") or {}).get(
                    "signature_guard_candidate_contracts"
                )
            )
            if isinstance(candidate, Mapping)
            and candidate.get("ready_for_candidate_knowledge_replay")
        )
        for item in worker_items
    )
    expected_signatures = sorted(
        {
            str(
                (
                    (item.get("embedding_signature_guard_replay") or {})
                    .get("embedding_signature_contract")
                    or {}
                ).get("embedding_signature")
                or ""
            )
            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_promotion_gate_replay"
    elif not execute and ready_count:
        next_action = "run_marketplace_embedding_signature_guard_replay_execute"
    elif execute and receipt_written_count:
        next_action = "run_marketplace_candidate_knowledge_replay"
    elif guard_failed_count:
        next_action = "repair_marketplace_embedding_signature_guard_replay_inputs"
    else:
        next_action = "refresh_marketplace_embedding_signature_guard_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,
        "signature_guard_candidate_count": signature_guard_candidate_count,
        "candidate_knowledge_ready_count": candidate_knowledge_ready_count,
        "expected_embedding_signatures": expected_signatures,
        "platforms": sorted({str(item.get("platform") or "unknown") for item in worker_items}),
        "network_call_performed": False,
        "model_call_performed": False,
        "embedding_generation_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(),
        "embedding_signature_guard_replay_version": (
            EMBEDDING_SIGNATURE_GUARD_REPLAY_VERSION
        ),
        "promotion_gate_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,
            "embedding_generation": 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,
            "embedding_signature_guard_replay_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_candidate_knowledge_replay": True,
        },
        "next_machine_action": next_action,
    }


__all__ = [
    "DEFAULT_OUTPUT_ROOT",
    "EMBEDDING_SIGNATURE_GUARD_REPLAY_VERSION",
    "POLICY",
    "run_pixelrag_marketplace_embedding_signature_guard_replay",
]
