#!/usr/bin/env python3
"""Replay one live HubSpot Contact through the normal inbound webhook.

This is a recovery tool for a Lead whose HubSpot mirror exists but whose
durable connector authority may be absent. It deliberately does not insert into
``_System_ConnectorInboundRemoteIDEvidence`` or ``_System_ConnectorRemoteLink``.
Instead, it verifies the Contact against HubSpot and delivers a signed ordinary
``contact.propertyChange`` webhook. The existing connector worker and Flow then
own reconciliation and durable-authorship creation.

The tool is single-candidate and dry-run by default. A current HubSpot record,
an exact Lead/Contact ID match, an email match, and a non-empty evidence
reference are all required before it will submit a webhook. Existing authority
does not need a special branch: the normal inbound path is idempotent. A missing
HubSpot record is a BLOCK, not an invitation to synthesize authority from the
CRM mirror.

The manifest is intentionally local and must contain exactly one candidate:

{
  "version": 1,
  "candidate": {
    "leadId": "...",
    "contactId": "...",
    "evidenceRef": "case or retained-source reference"
  }
}

Secrets are read only from environment variables and are never written into the
report. The report records hashes rather than contact values.
"""
from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode, urlsplit
from urllib.request import Request, urlopen


API_VERSION = "v63.0"
CONNECTOR_NAMESPACE = "hubspot-crm-connector"
CONNECTOR_WEBHOOK_PATH = f"/webhooks/{CONNECTOR_NAMESPACE}/webhook"
DEFAULT_TIMEOUT_SECONDS = 30
DEFAULT_POLL_TIMEOUT_SECONDS = 150
DEFAULT_POLL_INTERVAL_SECONDS = 5


class ReplayBlocked(RuntimeError):
    """A candidate did not meet the source-of-truth recovery contract."""


@dataclass(frozen=True)
class Candidate:
    lead_id: str
    contact_id: str
    evidence_ref: str


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--manifest", required=True, help="Local one-candidate JSON manifest.")
    parser.add_argument("--base-url", required=True, help="NexusCRM public base URL.")
    parser.add_argument(
        "--webhook-public-base-url",
        help="Public webhook base URL used for Host routing when connecting directly. Defaults to --base-url.",
    )
    parser.add_argument(
        "--webhook-connect-base-url",
        help="Optional direct connection URL for the signed webhook, such as http://127.0.0.1:3001."
        " The request is still signed for --webhook-public-base-url.",
    )
    parser.add_argument("--portal-id", required=True, help="Current HubSpot portal ID for the replay receipt.")
    parser.add_argument("--app-id", required=True, type=int, help="Current HubSpot app ID for the replay receipt.")
    parser.add_argument("--subscription-id", required=True, type=int, help="Current contact webhook subscription ID.")
    parser.add_argument("--hubspot-access-token-env", default="HUBSPOT_ACCESS_TOKEN")
    parser.add_argument("--hubspot-client-secret-env", default="HUBSPOT_CLIENT_SECRET")
    parser.add_argument("--nexus-username-env", default="NEXUS_ADMIN_USERNAME")
    parser.add_argument("--nexus-password-env", default="NEXUS_ADMIN_PASSWORD")
    parser.add_argument("--client-id", default="nexuscrm-ui")
    parser.add_argument("--client-secret", default="nexuscrm-ui-internal")
    parser.add_argument("--timeout-seconds", type=int, default=DEFAULT_TIMEOUT_SECONDS)
    parser.add_argument("--poll-timeout-seconds", type=int, default=DEFAULT_POLL_TIMEOUT_SECONDS)
    parser.add_argument("--poll-interval-seconds", type=int, default=DEFAULT_POLL_INTERVAL_SECONDS)
    parser.add_argument("--report", help="Optional JSON report path. Secrets and raw contact values are omitted.")
    parser.add_argument("--apply", action="store_true", help="Submit the verified replay. Default is dry-run.")
    parser.add_argument(
        "--confirm",
        default="",
        help="Required with --apply; must equal REPLAY_FROM_LIVE_HUBSPOT.",
    )
    return parser.parse_args()


def utc_now() -> dt.datetime:
    return dt.datetime.now(dt.timezone.utc)


def utc_timestamp() -> str:
    return utc_now().replace(microsecond=0).isoformat().replace("+00:00", "Z")


def sha256_text(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def normalized_email(value: object) -> str:
    return str(value or "").strip().lower()


def env_required(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise ReplayBlocked(f"required environment variable {name} is not set")
    return value


def load_candidate(path: Path) -> Candidate:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ReplayBlocked(f"cannot read manifest {path}: {exc}") from exc
    if not isinstance(payload, dict) or payload.get("version") != 1:
        raise ReplayBlocked("manifest must be an object with version=1")
    raw = payload.get("candidate")
    if not isinstance(raw, dict):
        raise ReplayBlocked("manifest must contain exactly one candidate object")
    candidate = Candidate(
        lead_id=str(raw.get("leadId") or "").strip(),
        contact_id=str(raw.get("contactId") or "").strip(),
        evidence_ref=str(raw.get("evidenceRef") or "").strip(),
    )
    if not candidate.lead_id or not candidate.contact_id or not candidate.evidence_ref:
        raise ReplayBlocked("candidate requires non-empty leadId, contactId, and evidenceRef")
    return candidate


def request_json(
    url: str,
    *,
    method: str = "GET",
    headers: dict[str, str] | None = None,
    payload: object | None = None,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
) -> tuple[int, dict[str, Any]]:
    data = None
    request_headers = dict(headers or {})
    if payload is not None:
        data = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        request_headers.setdefault("Content-Type", "application/json")
    request = Request(url, data=data, headers=request_headers, method=method)
    try:
        with urlopen(request, timeout=timeout) as response:
            raw = response.read().decode("utf-8")
            return response.status, json.loads(raw) if raw else {}
    except HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        try:
            body = json.loads(raw) if raw else {}
        except json.JSONDecodeError:
            body = {"raw": raw[:500]}
        return exc.code, body
    except URLError as exc:
        raise ReplayBlocked(f"{method} {url} could not connect: {exc}") from exc


def login(base_url: str, username: str, password: str, client_id: str, client_secret: str, timeout: int) -> str:
    payload = urlencode(
        {
            "grant_type": "password",
            "client_id": client_id,
            "client_secret": client_secret,
            "username": username,
            "password": password,
        }
    ).encode("utf-8")
    request = Request(
        f"{base_url.rstrip('/')}/services/oauth2/token",
        data=payload,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            body = json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        raise ReplayBlocked(f"NexusCRM login failed with HTTP {exc.code}") from exc
    token = str(body.get("access_token") or "").strip()
    if not token:
        raise ReplayBlocked("NexusCRM login returned no access token")
    return token


def soql_literal(value: str) -> str:
    return value.replace("\\", "\\\\").replace("'", "\\'")


def query_records(base_url: str, token: str, soql: str, timeout: int) -> list[dict[str, Any]]:
    url = f"{base_url.rstrip('/')}/services/data/{API_VERSION}/query?{urlencode({'q': soql})}"
    status, body = request_json(url, headers={"Authorization": f"Bearer {token}"}, timeout=timeout)
    if status != 200:
        raise ReplayBlocked(f"NexusCRM SOQL query returned HTTP {status}")
    records = body.get("records")
    if not isinstance(records, list):
        raise ReplayBlocked("NexusCRM SOQL response has no records array")
    return [record for record in records if isinstance(record, dict)]


def query_single_lead(base_url: str, token: str, lead_id: str, timeout: int) -> dict[str, Any]:
    records = query_records(
        base_url,
        token,
        "SELECT Id, Email, HubSpot_Id__c, IsConverted "
        f"FROM Lead WHERE Id = '{soql_literal(lead_id)}' LIMIT 2",
        timeout,
    )
    if len(records) != 1:
        raise ReplayBlocked(f"expected exactly one Lead for {lead_id}; got {len(records)}")
    return records[0]


def fetch_hubspot_contact(access_token: str, contact_id: str, timeout: int) -> dict[str, Any]:
    properties = "email,firstname,lastname,company"
    url = (
        f"https://api.hubapi.com/crm/v3/objects/contacts/{quote(contact_id, safe='')}?"
        f"properties={quote(properties, safe=',')}"
    )
    status, body = request_json(url, headers={"Authorization": f"Bearer {access_token}"}, timeout=timeout)
    if status == 404:
        raise ReplayBlocked(f"HubSpot contact {contact_id} is not available as an active record")
    if status != 200:
        raise ReplayBlocked(f"HubSpot contact {contact_id} query returned HTTP {status}")
    if str(body.get("id") or "").strip() != contact_id:
        raise ReplayBlocked("HubSpot contact response ID does not match the requested contact ID")
    properties_map = body.get("properties")
    if not isinstance(properties_map, dict) or not normalized_email(properties_map.get("email")):
        raise ReplayBlocked("HubSpot contact has no usable email for source verification")
    return body


def validate_candidate(candidate: Candidate, lead: dict[str, Any], hubspot_contact: dict[str, Any]) -> None:
    if bool(lead.get("IsConverted")):
        raise ReplayBlocked("candidate Lead is already converted")
    if str(lead.get("HubSpot_Id__c") or "").strip() != candidate.contact_id:
        raise ReplayBlocked("candidate Lead mirror does not match the HubSpot contact ID")
    hubspot_properties = hubspot_contact.get("properties")
    if not isinstance(hubspot_properties, dict):
        raise ReplayBlocked("HubSpot contact response has invalid properties")
    if normalized_email(lead.get("Email")) != normalized_email(hubspot_properties.get("email")):
        raise ReplayBlocked("candidate Lead email does not match the live HubSpot contact email")
def replay_event(args: argparse.Namespace, candidate: Candidate, hubspot_contact: dict[str, Any]) -> dict[str, Any]:
    hubspot_properties = hubspot_contact["properties"]
    event = {
        "portalId": str(args.portal_id),
        "appId": args.app_id,
        "subscriptionId": args.subscription_id,
        "eventId": f"nexuscrm-authoritative-replay-{candidate.contact_id}-{int(time.time() * 1000)}",
        "occurredAt": str(int(time.time() * 1000)),
        "subscriptionType": "contact.propertyChange",
        "attemptNumber": 0,
        "objectId": candidate.contact_id,
        "propertyName": "email",
        "propertyValue": str(hubspot_properties["email"]),
        "changeSource": "INTEGRATION",
    }
    return event


def sign_webhook(body: bytes, secret: str) -> str:
    return hashlib.sha256(secret.encode("utf-8") + body).hexdigest()


def webhook_urls(args: argparse.Namespace) -> tuple[str, str]:
    public_base = str(args.webhook_public_base_url or args.base_url).rstrip("/")
    connect_base = str(args.webhook_connect_base_url or public_base).rstrip("/")
    if not public_base or not connect_base:
        raise ReplayBlocked("webhook public and connection base URLs must be non-empty")
    return f"{public_base}{CONNECTOR_WEBHOOK_PATH}", f"{connect_base}{CONNECTOR_WEBHOOK_PATH}"


def submit_replay(args: argparse.Namespace, event: dict[str, Any], client_secret: str) -> tuple[int, dict[str, Any]]:
    public_url, connect_url = webhook_urls(args)
    payload = [event]
    body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
    signature = sign_webhook(body, client_secret)
    headers = {
        "Content-Type": "application/json",
        "X-HubSpot-Signature": signature,
    }
    if connect_url != public_url:
        public_host = urlsplit(public_url).netloc
        if public_host:
            headers["Host"] = public_host
    return request_json(
        connect_url,
        method="POST",
        headers=headers,
        payload=payload,
        timeout=args.timeout_seconds,
    )


def get_ingress_receipt(base_url: str, token: str, receipt_id: str, timeout: int) -> dict[str, Any]:
    url = f"{base_url.rstrip('/')}/api/connectors/ingress-receipts/{quote(receipt_id, safe='')}"
    status, body = request_json(url, headers={"Authorization": f"Bearer {token}"}, timeout=timeout)
    if status != 200:
        raise ReplayBlocked(f"ingress receipt {receipt_id} query returned HTTP {status}")
    receipt = body.get("receipt")
    if not isinstance(receipt, dict):
        raise ReplayBlocked("ingress receipt response has no receipt object")
    return receipt


def wait_for_durable_ingress(
    args: argparse.Namespace,
    nexus_token: str,
    response: dict[str, Any],
) -> tuple[str, list[str]]:
    event_ids = response.get("eventIds")
    if isinstance(event_ids, list) and event_ids:
        ids = [str(value) for value in event_ids if str(value).strip()]
        if ids:
            return "inline_event", ids

    receipt_id = str(response.get("receiptId") or "").strip()
    if not receipt_id:
        raise ReplayBlocked("accepted webhook returned neither receiptId nor eventIds")

    deadline = time.monotonic() + args.poll_timeout_seconds
    while time.monotonic() < deadline:
        receipt = get_ingress_receipt(args.base_url, nexus_token, receipt_id, args.timeout_seconds)
        status = str(receipt.get("status") or "").strip().lower()
        if status == "enqueued":
            values = receipt.get("enqueuedEventIds")
            ids = [str(value) for value in values] if isinstance(values, list) else []
            if not ids:
                raise ReplayBlocked(f"ingress receipt {receipt_id} is enqueued without event IDs")
            return "verified_receipt", ids
        if status in {"dead", "config_terminal"}:
            error_class = str(receipt.get("lastErrorClass") or "unknown").strip()
            raise ReplayBlocked(f"ingress receipt {receipt_id} reached {status} ({error_class})")
        time.sleep(args.poll_interval_seconds)
    raise ReplayBlocked(f"ingress receipt {receipt_id} did not enqueue before timeout")


def candidate_report(candidate: Candidate, lead: dict[str, Any], hubspot_contact: dict[str, Any]) -> dict[str, Any]:
    properties = hubspot_contact["properties"]
    return {
        "leadId": candidate.lead_id,
        "contactId": candidate.contact_id,
        "evidenceRef": candidate.evidence_ref,
        "leadEmailSha256": sha256_text(normalized_email(lead.get("Email"))),
        "hubspotEmailSha256": sha256_text(normalized_email(properties.get("email"))),
        "hubspotContactHash": sha256_text(str(hubspot_contact.get("id") or "")),
    }


def write_report(path: str | None, report: dict[str, Any]) -> None:
    if not path:
        return
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def run(args: argparse.Namespace) -> dict[str, Any]:
    if args.apply and args.confirm != "REPLAY_FROM_LIVE_HUBSPOT":
        raise ReplayBlocked("--apply requires --confirm REPLAY_FROM_LIVE_HUBSPOT")
    if args.poll_timeout_seconds <= 0 or args.poll_interval_seconds <= 0:
        raise ReplayBlocked("poll timeout and interval must be positive")

    candidate = load_candidate(Path(args.manifest))
    hubspot_token = env_required(args.hubspot_access_token_env)
    nexus_username = env_required(args.nexus_username_env)
    nexus_password = env_required(args.nexus_password_env)
    nexus_token = login(args.base_url, nexus_username, nexus_password, args.client_id, args.client_secret, args.timeout_seconds)
    lead = query_single_lead(args.base_url, nexus_token, candidate.lead_id, args.timeout_seconds)
    hubspot_contact = fetch_hubspot_contact(hubspot_token, candidate.contact_id, args.timeout_seconds)
    validate_candidate(candidate, lead, hubspot_contact)

    report: dict[str, Any] = {
        "generatedAt": utc_timestamp(),
        "status": "DRY_RUN_PASS" if not args.apply else "PREFLIGHT_PASS",
        "mode": "apply" if args.apply else "dry-run",
        "candidate": candidate_report(candidate, lead, hubspot_contact),
        "directSystemTableWrite": False,
        "source": "live_hubspot_contact_get",
    }
    if not args.apply:
        return report

    client_secret = env_required(args.hubspot_client_secret_env)
    event = replay_event(args, candidate, hubspot_contact)
    status, response = submit_replay(args, event, client_secret)
    report["webhookStatus"] = status
    report["replayEventId"] = event["eventId"]
    report["replayEventHash"] = sha256_text(json.dumps(event, sort_keys=True, separators=(",", ":")))
    if status != 202:
        report["status"] = "WEBHOOK_REJECTED"
        report["webhookResponseKeys"] = sorted(str(key) for key in response.keys())
        return report

    durable_kind, event_ids = wait_for_durable_ingress(args, nexus_token, response)
    report["status"] = "PASS"
    report["completionBoundary"] = "durable_ingress_enqueued"
    report["durableIngressKind"] = durable_kind
    report["enqueuedEventIds"] = event_ids
    return report


def main() -> int:
    args = parse_args()
    try:
        report = run(args)
    except ReplayBlocked as exc:
        report = {
            "generatedAt": utc_timestamp(),
            "status": "BLOCKED",
            "mode": "apply" if args.apply else "dry-run",
            "reason": str(exc),
            "directSystemTableWrite": False,
        }
    write_report(args.report, report)
    print(json.dumps(report, sort_keys=True))
    return 0 if report["status"] in {"DRY_RUN_PASS", "PASS"} else 2


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