import hashlib
import hmac
import json
import re
import time
from collections.abc import Callable, Sequence
from datetime import datetime
from typing import TypedDict


SOURCE = "https://elicitra.eigen.rest"
LIFECYCLE_DOCS_BASE = f"{SOURCE}/developers/v5/schemas/webhooks"
AUTOMATION_DOCS_BASE = f"{SOURCE}/developers/v6/schemas/webhooks"
EVENT_TYPES = (
    "com.elicitra.interaction.ended.v1",
    "com.elicitra.analysis.completed.v1",
    "com.elicitra.analysis.failed.v1",
    "com.elicitra.automation.triggered.v1",
)


class WebhookAcceptance(TypedDict):
    status: int
    accepted: bool
    duplicate: bool
    event_id: str | None
    challenge: str | None
    error: str | None


PersistEventOnce = Callable[[str, bytes], bool]
# Compile lifecycle schemas from `/developers/v5/` and the automation schema
# from `/developers/v6/`, then dispatch by event type and dataschema.
ValidateEventData = Callable[[str, str, object], bool]


def verify_elicitra_signature(
    raw_body: bytes,
    header: str,
    secrets: Sequence[str],
    now_seconds: float | None = None,
) -> bool:
    parts = [part.strip() for part in header.split(",")]
    timestamp = next((part[2:] for part in parts if part.startswith("t=")), None)
    signatures = [
        part[3:]
        for part in parts
        if part.startswith("v1=") and re.fullmatch(r"[0-9a-fA-F]{64}", part[3:])
    ]
    if (
        timestamp is None
        or len(timestamp) > 16
        or not timestamp.isdigit()
        or not signatures
    ):
        return False
    try:
        timestamp_number = int(timestamp)
    except (ValueError, OverflowError):
        return False
    now = time.time() if now_seconds is None else now_seconds
    if abs(now - timestamp_number) > 300:
        return False

    signed = timestamp.encode() + b"." + raw_body
    return any(
        hmac.compare_digest(
            hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest(), candidate
        )
        for secret in secrets
        for candidate in signatures
    )


def _invalid(error: str, status: int) -> WebhookAcceptance:
    return {
        "status": status,
        "accepted": False,
        "duplicate": False,
        "event_id": None,
        "challenge": None,
        "error": error,
    }


def _parse_cloud_event(
    value: object, validate_event_data: ValidateEventData
) -> dict[str, object] | None:
    if not isinstance(value, dict):
        return None
    required = {
        "specversion",
        "id",
        "source",
        "type",
        "subject",
        "time",
        "datacontenttype",
        "dataschema",
        "data",
    }
    if set(value) != required:
        return None

    event_type = value.get("type")
    event_id = value.get("id")
    subject = value.get("subject")
    event_time = value.get("time")
    dataschema = value.get("dataschema")
    data = value.get("data")
    if not isinstance(event_type, str) or event_type not in EVENT_TYPES:
        return None
    expected_sequence = (
        1
        if event_type == EVENT_TYPES[0]
        else 2
        if event_type in EVENT_TYPES[1:3]
        else None
    )
    if (
        value.get("specversion") != "1.0"
        or not isinstance(event_id, str)
        or not 1 <= len(event_id) <= 256
        or value.get("source") != SOURCE
        or not isinstance(subject, str)
        or re.fullmatch(r"interactions/[A-Za-z0-9_-]{1,256}", subject) is None
        or not isinstance(event_time, str)
        or re.fullmatch(
            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})",
            event_time,
        )
        is None
        or value.get("datacontenttype") != "application/json"
        or not isinstance(dataschema, str)
        or dataschema
        != f"{AUTOMATION_DOCS_BASE if event_type == EVENT_TYPES[3] else LIFECYCLE_DOCS_BASE}/{event_type}.schema.json"
        or not isinstance(data, dict)
        or (expected_sequence is None and "lifecycleSequence" in data)
        or (
            expected_sequence is not None
            and data.get("lifecycleSequence") != expected_sequence
        )
    ):
        return None
    try:
        datetime.fromisoformat(event_time.replace("Z", "+00:00"))
    except ValueError:
        return None
    if not validate_event_data(event_type, dataschema, data):
        return None
    return value


def accept_elicitra_webhook(
    *,
    raw_body: bytes,
    signature_header: str,
    secrets: Sequence[str],
    validate_event_data: ValidateEventData,
    persist_event_once: PersistEventOnce,
    expose_challenge: Callable[[str], None] | None = None,
    now_seconds: float | None = None,
) -> WebhookAcceptance:
    # Authenticate the exact bytes before parsing or changing the body.
    if not verify_elicitra_signature(raw_body, signature_header, secrets, now_seconds):
        return _invalid("invalid_signature", 401)

    try:
        value = json.loads(raw_body)
    except (json.JSONDecodeError, UnicodeDecodeError):
        return _invalid("invalid_cloudevent", 400)

    # Endpoint verification is a signed control message, not a CloudEvent.
    if (
        isinstance(value, dict)
        and set(value) == {"challenge"}
        and isinstance(value["challenge"], str)
        and 16 <= len(value["challenge"]) <= 256
    ):
        if expose_challenge is not None:
            expose_challenge(value["challenge"])
        return {
            "status": 200,
            "accepted": True,
            "duplicate": False,
            "event_id": None,
            "challenge": value["challenge"],
            "error": None,
        }

    event = _parse_cloud_event(value, validate_event_data)
    if event is None:
        return _invalid("invalid_cloudevent", 400)

    event_id = event["id"]
    assert isinstance(event_id, str)
    # This adapter must be one atomic insert protected by a unique event-id
    # constraint. Do not implement it as SELECT followed by INSERT.
    inserted = persist_event_once(event_id, raw_body)
    return {
        "status": 202 if inserted else 200,
        "accepted": True,
        "duplicate": not inserted,
        "event_id": event_id,
        "challenge": None,
        "error": None,
    }


def _sign(raw_body: bytes, timestamp: str, secret: str) -> str:
    return hmac.new(
        secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256
    ).hexdigest()


if __name__ == "__main__":
    test_secret = "whsec_test_only"
    rotated_secret = "whsec_rotated_test_only"
    test_timestamp = "1785499200"
    event_type = EVENT_TYPES[0]
    test_data = {
        "lifecycleSequence": 1,
        "organization": {"id": "org_01"},
        "interaction": {
            "id": "call_01",
            "channel": "phone",
            "outcome": "completed",
            "reasonCode": "respondent_ended",
            "endedAt": "2026-07-31T12:00:00.000Z",
        },
        "campaign": None,
        "scenario": {
            "id": "scenario_01",
            "slug": "customer_research",
            "name": "Customer research",
            "revision": 1,
        },
        "respondent": {"name": None, "email": None, "phone": None},
        "consents": [],
        "truncation": {"truncated": False, "omittedFields": []},
    }
    test_event = {
        "specversion": "1.0",
        "id": "evt_test",
        "source": SOURCE,
        "type": event_type,
        "subject": "interactions/call_01",
        "time": "2026-07-31T12:00:00.000Z",
        "datacontenttype": "application/json",
        "dataschema": f"{AUTOMATION_DOCS_BASE if event_type == EVENT_TYPES[3] else LIFECYCLE_DOCS_BASE}/{event_type}.schema.json",
        "data": test_data,
    }
    test_body = json.dumps(test_event, separators=(",", ":")).encode()
    test_signature = _sign(test_body, test_timestamp, test_secret)
    inbox: dict[str, bytes] = {}

    def persist_event_once(event_id: str, raw_body: bytes) -> bool:
        # Test double only. Production storage must enforce uniqueness atomically.
        if event_id in inbox:
            return False
        inbox[event_id] = raw_body
        return True

    # Test double only. Production code must compile and run the pinned JSON
    # Schema 2020-12 data schema selected by type and dataschema.
    def validate_event_data(
        candidate_type: str, candidate_schema: str, candidate_data: object
    ) -> bool:
        return (
            candidate_type == event_type
            and candidate_schema == test_event["dataschema"]
            and candidate_data == test_data
        )

    accepted = accept_elicitra_webhook(
        raw_body=test_body,
        signature_header=f"t={test_timestamp},v1={test_signature}",
        secrets=[test_secret],
        validate_event_data=validate_event_data,
        persist_event_once=persist_event_once,
        now_seconds=int(test_timestamp),
    )
    assert accepted == {
        "status": 202,
        "accepted": True,
        "duplicate": False,
        "event_id": "evt_test",
        "challenge": None,
        "error": None,
    }

    duplicate = accept_elicitra_webhook(
        raw_body=test_body,
        signature_header=f"t={test_timestamp},v1={test_signature}",
        secrets=[test_secret],
        validate_event_data=validate_event_data,
        persist_event_once=persist_event_once,
        now_seconds=int(test_timestamp),
    )
    assert duplicate["status"] == 200 and duplicate["duplicate"]
    assert len(inbox) == 1

    challenge_body = json.dumps(
        {"challenge": "0123456789abcdef0123456789abcdef"}, separators=(",", ":")
    ).encode()
    observed: list[str] = []
    challenge = accept_elicitra_webhook(
        raw_body=challenge_body,
        signature_header=(
            f"t={test_timestamp},v1={_sign(challenge_body, test_timestamp, test_secret)}"
        ),
        secrets=[test_secret],
        validate_event_data=validate_event_data,
        persist_event_once=persist_event_once,
        expose_challenge=observed.append,
        now_seconds=int(test_timestamp),
    )
    assert challenge["status"] == 200
    assert observed == ["0123456789abcdef0123456789abcdef"]
    assert len(inbox) == 1

    tampered = accept_elicitra_webhook(
        raw_body=test_body + b" ",
        signature_header=f"t={test_timestamp},v1={test_signature}",
        secrets=[test_secret],
        validate_event_data=validate_event_data,
        persist_event_once=persist_event_once,
        now_seconds=int(test_timestamp),
    )
    assert tampered["status"] == 401 and len(inbox) == 1

    assert not verify_elicitra_signature(
        test_body,
        f"t={test_timestamp},v1={test_signature}",
        [test_secret],
        int(test_timestamp) + 301,
    )
    assert not verify_elicitra_signature(
        test_body,
        f"t={'1' * 10_000},v1={test_signature}",
        [test_secret],
        int(test_timestamp),
    )

    rotated_event = {**test_event, "id": "evt_rotated"}
    rotated_body = json.dumps(rotated_event, separators=(",", ":")).encode()
    old_signature = _sign(rotated_body, test_timestamp, test_secret)
    new_signature = _sign(rotated_body, test_timestamp, rotated_secret)
    rotated = accept_elicitra_webhook(
        raw_body=rotated_body,
        signature_header=(
            f"t={test_timestamp},v1={old_signature},v1={new_signature}"
        ),
        secrets=[rotated_secret],
        validate_event_data=validate_event_data,
        persist_event_once=persist_event_once,
        now_seconds=int(test_timestamp),
    )
    assert rotated["status"] == 202 and len(inbox) == 2

    malformed_body = b"{not-json"
    malformed = accept_elicitra_webhook(
        raw_body=malformed_body,
        signature_header=(
            f"t={test_timestamp},v1={_sign(malformed_body, test_timestamp, test_secret)}"
        ),
        secrets=[test_secret],
        validate_event_data=validate_event_data,
        persist_event_once=persist_event_once,
        now_seconds=int(test_timestamp),
    )
    assert malformed["status"] == 400 and len(inbox) == 2