"""7K Forms automation/webhook composition: trigger vs internal forward."""
from __future__ import annotations

from datetime import UTC, datetime
from pathlib import Path

from src.domain.connectors.invoke_capabilities import (
    CAPABILITY_CONVERSION_PATTERN,
    FORMS_FAMILY_PLAN,
    INVOKE_CAPABILITIES,
)
from src.domain.forms.actions import (
    FormCapabilityInvokeResult,
    FormPostSubmitAction,
    FormPostSubmitContext,
    FormPostSubmitPlan,
    RecordingCapabilityInvoker,
    action_idempotency_key,
    contact_compatible_plan,
    execute_post_submit_actions,
    validate_post_submit_plan,
)
from src.domain.forms.automation import (
    CONTACT_COMPATIBLE_AUTOMATION_MAPPING,
    FORMS_AUTOMATION_COMPOSITION,
    GENERIC_AUTOMATION_ACTION,
    automation_workflow_action,
    build_automation_trigger_payload,
    validate_automation_mapping,
)
from src.domain.forms.canonical import (
    FormDefinition,
    FormField,
    FormFieldOption,
    FormFileRef,
    FormSubmission,
    contact_block_compatible_definition,
)
from src.domain.forms.email import email_send_action
from src.domain.forms.forward import (
    FORMS_HTTP_FORWARD_COMPOSITION,
    SSRF_TRANSPORT,
    http_forward_action,
    validate_forward_target,
    validate_http_forward_mapping,
)
from src.domain.forms.persistence import (
    FormPersistenceContext,
    FormPersistenceError,
    InMemoryFormSubmissionStore,
    get_form_submission,
)
from src.domain.forms.runtime import accept_form_submission

_FIXED_AT = datetime(2026, 9, 13, 22, 0, tzinfo=UTC)
_FSB_A = "fsb_" + ("a" * 32)
_FSB_B = "fsb_" + ("b" * 32)
_FILE_1 = "file_" + ("1" * 32)
_CTX = FormPostSubmitContext(project_id="prj_a")
_SAFE_TARGET = "https://example.com/webhook"


class BoomStore:
    def persist(self, *args, **kwargs):
        raise FormPersistenceError("submission_persistence_failed")

    def get(self, *args, **kwargs):
        return None


def _accept(definition, raw, submission_id=_FSB_A):
    result = accept_form_submission(
        definition,
        raw,
        clock=lambda: _FIXED_AT,
        submission_id_factory=lambda: submission_id,
    )
    assert result.accepted is True
    return result.submission


def _fields(**extra: bool) -> FormDefinition:
    fields = [
        FormField(key="email", type="email", required=False),
        FormField(key="name", type="text", required=False),
        FormField(key="message", type="textarea", required=False),
        FormField(key="phone", type="phone", required=False),
        FormField(key="opt_in", type="checkbox", required=False),
        FormField(key="budget", type="number", required=False),
        FormField(key="note", type="text", required=False),
        FormField(key="callback_url", type="text", required=False),
        FormField(key="workflow", type="text", required=False),
        FormField(key="adapter", type="text", required=False),
    ]
    if extra.get("file"):
        fields.append(FormField(key="resume", type="file", required=False))
    if extra.get("multi"):
        fields.append(
            FormField(
                key="tags",
                type="multi_select",
                required=False,
                options=[FormFieldOption(value="a", label="A")],
            )
        )
    return FormDefinition(key="lead", fields=fields)


def test_7k_composition_frozen() -> None:
    assert FORMS_AUTOMATION_COMPOSITION["status"] == "frozen"
    assert FORMS_AUTOMATION_COMPOSITION["generic_action"] == GENERIC_AUTOMATION_ACTION
    assert FORMS_AUTOMATION_COMPOSITION["execution_domain"] == "C"
    assert FORMS_AUTOMATION_COMPOSITION["workflow_identity"] == "logical_alias"
    assert FORMS_AUTOMATION_COMPOSITION["resolver_in_forms"] is False
    assert FORMS_AUTOMATION_COMPOSITION["file_policy"] == "disallow"
    assert FORMS_HTTP_FORWARD_COMPOSITION["action"] == "internal.http_forward"
    assert FORMS_HTTP_FORWARD_COMPOSITION["not_canonical_capability"] is True
    assert FORMS_HTTP_FORWARD_COMPOSITION["cloud_request_type"] == "form.forward"
    assert FORMS_HTTP_FORWARD_COMPOSITION["method"] == "POST"
    assert FORMS_HTTP_FORWARD_COMPOSITION["ssrf_transport"] == SSRF_TRANSPORT
    assert "form.forward" not in INVOKE_CAPABILITIES
    assert "webhook.send" not in INVOKE_CAPABILITIES
    assert GENERIC_AUTOMATION_ACTION in INVOKE_CAPABILITIES
    assert CAPABILITY_CONVERSION_PATTERN["phase_7_status"] == "complete"
    assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == (
        "9_app_block_modernization"
    )
    assert CAPABILITY_CONVERSION_PATTERN["opened_families"]["forms.*"] == (
        "composition_integration"
    )
    assert FORMS_FAMILY_PLAN["status"] == "complete"
    assert FORMS_FAMILY_PLAN["automation_webhook_composition_frozen"] is True
    assert FORMS_FAMILY_PLAN["slices"][10] == "7k_automation_webhook_composition"
    assert FORMS_FAMILY_PLAN["next_slice"] == "9_app_block_modernization"
    assert "workflow_status" not in FormSubmission.model_fields
    auto_src = Path("src/domain/forms/automation.py").read_text(encoding="utf-8").lower()
    fwd_src = Path("src/domain/forms/forward.py").read_text(encoding="utf-8").lower()
    assert "zapier" not in auto_src
    assert "zapier" not in fwd_src
    assert "if make" not in auto_src
    assert "urllib.request" not in fwd_src
    assert "http.client" not in fwd_src


def test_explicit_scalar_mapping_preserves_missing_null_false_zero() -> None:
    definition = _fields()
    action = automation_workflow_action(
        mapping={
            "email": "email",
            "name": "name",
            "opt_in": "opt_in",
            "amount": "budget",
            "note": "note",
        }
    )
    built = build_automation_trigger_payload(
        action,
        {"name": "Ada", "opt_in": False, "budget": 0, "note": None},
    )
    assert built.payload is not None
    assert built.payload["workflow"] == "lead-intake"
    assert "email" not in built.payload["input"]
    assert built.payload["input"]["name"] == "Ada"
    assert built.payload["input"]["opt_in"] is False
    assert built.payload["input"]["amount"] == 0
    assert built.payload["input"]["note"] is None
    accepted = _accept(
        definition, {"email": "ada@example.com", "opt_in": False, "budget": 0}
    )
    invoker = RecordingCapabilityInvoker()
    execute_post_submit_actions(
        accepted,
        FormPostSubmitPlan(actions=(action,)),
        context=_CTX,
        invoker=invoker,
        definition=definition,
    )
    assert invoker.calls[0]["payload"]["input"]["email"] == "ada@example.com"
    assert invoker.calls[0]["payload"]["input"]["opt_in"] is False
    assert invoker.calls[0]["payload"]["input"]["amount"] == 0


def test_make_n8n_see_identical_forms_payload() -> None:
    definition = _fields()
    accepted = _accept(definition, {"email": "ada@example.com", "message": "Hi"})
    plan = FormPostSubmitPlan(actions=(automation_workflow_action(),))
    make = RecordingCapabilityInvoker()
    n8n = RecordingCapabilityInvoker()
    execute_post_submit_actions(
        accepted, plan, context=_CTX, invoker=make, definition=definition
    )
    execute_post_submit_actions(
        accepted, plan, context=_CTX, invoker=n8n, definition=definition
    )
    assert make.calls[0]["payload"] == n8n.calls[0]["payload"]
    assert "make" not in str(make.calls[0]["payload"]).lower()
    assert "n8n" not in str(make.calls[0]["payload"]).lower()
    assert "webhook_url" not in str(make.calls[0]["payload"])


def test_workflow_unbound_and_binding_ambiguous_pass_through() -> None:
    definition = contact_block_compatible_definition()
    accepted = _accept(
        definition,
        {"name": "Ada", "email": "ada@example.com", "message": "Hi"},
    )
    unbound = RecordingCapabilityInvoker(
        {
            "automation.workflow.trigger": FormCapabilityInvokeResult(
                status="failed",
                error_code="workflow_unbound",
            )
        }
    )
    result = execute_post_submit_actions(
        accepted,
        contact_compatible_plan(),
        context=_CTX,
        invoker=unbound,
        definition=definition,
    )
    by_key = {item.action_key: item for item in result.actions}
    assert by_key["notify_automation"].error_code == "workflow_unbound"
    assert by_key["sync_crm"].status == "succeeded"
    assert by_key["send_confirmation"].status == "succeeded"
    ambiguous = RecordingCapabilityInvoker(
        {
            "automation.workflow.trigger": FormCapabilityInvokeResult(
                status="failed",
                error_code="binding_ambiguous",
            )
        }
    )
    hinted = FormPostSubmitPlan(
        actions=(
            automation_workflow_action(provider_hint="make_com"),
            http_forward_action(target=_SAFE_TARGET, mapping={"email": "email"}),
        )
    )
    later = execute_post_submit_actions(
        accepted, hinted, context=_CTX, invoker=ambiguous, definition=definition
    )
    assert later.actions[0].error_code == "binding_ambiguous"
    assert later.actions[1].status == "succeeded"
    assert later.actions[1].action_type == "internal.http_forward"


def test_user_controlled_routing_rejected() -> None:
    definition = _fields()
    workflow = automation_workflow_action(
        mapping={"workflow": "workflow", "email": "email"}
    )
    assert "routing_from_form_values" in validate_automation_mapping(workflow, definition)
    hint = automation_workflow_action(
        mapping={"provider_hint": "adapter", "email": "email"}
    )
    assert "routing_from_form_values" in validate_automation_mapping(hint, definition)
    nested = automation_workflow_action(mapping={"customer.email": "email"})
    assert "nested_automation_target" in validate_automation_mapping(nested, definition)
    mapped_target_bad = FormPostSubmitAction(
        key="forward",
        action_type="internal.http_forward",
        mapping={"target": "callback_url", "email": "email"},
        constants={"target": _SAFE_TARGET},
    )
    assert "routing_from_form_values" in validate_http_forward_mapping(
        mapped_target_bad, definition
    )
    mapped_target = FormPostSubmitAction(
        key="forward",
        action_type="internal.http_forward",
        mapping={"email": "email"},
        constants={"target": "https://example.com/x"},
    )
    assert validate_http_forward_mapping(mapped_target, definition) == ()


def test_file_and_anti_abuse_sources_rejected() -> None:
    definition = _fields(file=True, multi=True)
    resume = automation_workflow_action(mapping={"resume": "resume", "email": "email"})
    assert "unsupported_automation_file_mapping" in validate_automation_mapping(
        resume, definition
    )
    runtime = build_automation_trigger_payload(
        resume,
        {"email": "ada@example.com", "resume": FormFileRef(file_id=_FILE_1)},
    )
    assert runtime.error_code == "unsupported_automation_file_mapping"
    tags = automation_workflow_action(mapping={"tags": "tags", "email": "email"})
    assert "unsupported_automation_source_type" in validate_automation_mapping(
        tags, definition
    )
    honeypot = automation_workflow_action(mapping={"email": "turnstile_token"})
    assert "unknown_source_field" in validate_automation_mapping(honeypot, definition)
    file_url = http_forward_action(target=_SAFE_TARGET, mapping={"resume": "resume"})
    assert "unsupported_forward_file_mapping" in validate_http_forward_mapping(
        file_url, definition
    )


def test_http_forward_trusted_target_and_invalid_targets() -> None:
    definition = _fields()
    accepted = _accept(definition, {"email": "ada@example.com", "message": "Hi"})
    action = http_forward_action(
        target=_SAFE_TARGET,
        mapping={"email": "email", "message": "message"},
    )
    invoker = RecordingCapabilityInvoker()
    result = execute_post_submit_actions(
        accepted,
        FormPostSubmitPlan(actions=(action,)),
        context=_CTX,
        invoker=invoker,
        definition=definition,
    )
    assert result.actions[0].status == "succeeded"
    assert invoker.calls[0]["capability"] == "internal.http_forward"
    assert invoker.calls[0]["payload"] == {
        "target_url": _SAFE_TARGET,
        "method": "POST",
        "fields": {"email": "ada@example.com", "message": "Hi"},
    }
    assert validate_forward_target("http://127.0.0.1/") == "forward_target_invalid"
    assert validate_forward_target("http://localhost/") == "forward_target_invalid"
    assert validate_forward_target("http://169.254.169.254/") == "forward_target_invalid"
    assert validate_forward_target("ftp://example.com/") == "forward_target_invalid"
    private = http_forward_action(
        target="http://127.0.0.1/hook", mapping={"email": "email"}
    )
    assert "forward_target_invalid" in validate_http_forward_mapping(private, definition)
    blocked = execute_post_submit_actions(
        accepted,
        FormPostSubmitPlan(actions=(private,)),
        context=_CTX,
        invoker=RecordingCapabilityInvoker(),
        definition=definition,
    )
    assert blocked.plan_error == "forward_target_invalid"


def test_http_forward_failure_does_not_stop_later_actions() -> None:
    definition = _fields()
    accepted = _accept(definition, {"email": "ada@example.com"})
    plan = FormPostSubmitPlan(
        actions=(
            http_forward_action(target=_SAFE_TARGET, mapping={"email": "email"}),
            automation_workflow_action(mapping={"email": "email"}),
        )
    )
    invoker = RecordingCapabilityInvoker(
        {
            "internal.http_forward": FormCapabilityInvokeResult(
                status="failed",
                error_code="forward_failed",
            )
        }
    )
    result = execute_post_submit_actions(
        accepted, plan, context=_CTX, invoker=invoker, definition=definition
    )
    assert result.actions[0].error_code == "forward_failed"
    assert result.actions[1].status == "succeeded"
    assert [call["capability"] for call in invoker.calls] == [
        "internal.http_forward",
        "automation.workflow.trigger",
    ]


def test_two_submissions_two_automation_identities() -> None:
    definition = _fields()
    plan = FormPostSubmitPlan(
        actions=(
            automation_workflow_action(key="trigger_crm_workflow"),
            automation_workflow_action(
                key="trigger_marketing_workflow",
                workflow="marketing-nurture",
            ),
        )
    )
    invoker = RecordingCapabilityInvoker()
    for submission_id in (_FSB_A, _FSB_B):
        accepted = _accept(
            definition,
            {"email": "jane@example.com", "message": "Hi"},
            submission_id=submission_id,
        )
        execute_post_submit_actions(
            accepted, plan, context=_CTX, invoker=invoker, definition=definition
        )
    keys = [call["idempotency_key"] for call in invoker.calls]
    assert action_idempotency_key(_FSB_A, "trigger_crm_workflow") in keys
    assert action_idempotency_key(_FSB_A, "trigger_marketing_workflow") in keys
    assert keys.count(action_idempotency_key(_FSB_A, "trigger_crm_workflow")) == 1
    assert keys.count(action_idempotency_key(_FSB_B, "trigger_crm_workflow")) == 1


def test_persist_first_failure_invokes_neither() -> None:
    definition = contact_block_compatible_definition()
    accepted = _accept(
        definition,
        {"name": "Ada", "email": "ada@example.com", "message": "Hi"},
    )
    invoker = RecordingCapabilityInvoker()
    result = execute_post_submit_actions(
        accepted,
        FormPostSubmitPlan(
            actions=(
                automation_workflow_action(),
                http_forward_action(target=_SAFE_TARGET, mapping={"email": "email"}),
            )
        ),
        context=_CTX,
        invoker=invoker,
        definition=definition,
        store=BoomStore(),
        persist_first=True,
    )
    assert result.plan_error == "persistence_failed"
    assert invoker.calls == []


def test_automation_failure_does_not_write_forms_state() -> None:
    definition = contact_block_compatible_definition()
    accepted = _accept(
        definition,
        {"name": "Ada", "email": "ada@example.com", "message": "Hi"},
    )
    store = InMemoryFormSubmissionStore()
    invoker = RecordingCapabilityInvoker(
        {
            "automation.workflow.trigger": FormCapabilityInvokeResult(
                status="failed",
                error_code="workflow_unbound",
            )
        }
    )
    result = execute_post_submit_actions(
        accepted,
        contact_compatible_plan(),
        context=_CTX,
        invoker=invoker,
        definition=definition,
        store=store,
        persist_first=True,
    )
    row = get_form_submission(store, FormPersistenceContext(project_id="prj_a"), _FSB_A)
    assert row is not None
    assert "execution_ref" not in row.accepted.submission.values
    assert result.actions[2].error_code == "workflow_unbound"


def test_contact_compatible_plan_automation_is_logical() -> None:
    plan = contact_compatible_plan()
    auto = plan.actions[2]
    assert auto.action_type == "automation.workflow.trigger"
    assert auto.mapping == CONTACT_COMPATIBLE_AUTOMATION_MAPPING
    assert auto.constants == {"workflow": "lead-intake"}
    definition = contact_block_compatible_definition()
    assert validate_post_submit_plan(plan, definition) == ()
    accepted = _accept(
        definition,
        {"name": "Ada", "email": "ada@example.com", "phone": "555", "message": "Hi"},
    )
    invoker = RecordingCapabilityInvoker()
    execute_post_submit_actions(
        accepted, plan, context=_CTX, invoker=invoker, definition=definition
    )
    payload = invoker.calls[2]["payload"]
    assert payload["workflow"] == "lead-intake"
    assert payload["input"] == {
        "email": "ada@example.com",
        "phone": "555",
        "message": "Hi",
    }
    assert "name" not in payload["input"]


def test_queued_execution_ref_is_opaque() -> None:
    definition = _fields()
    accepted = _accept(definition, {"email": "ada@example.com"})
    invoker = RecordingCapabilityInvoker(
        {
            "automation.workflow.trigger": FormCapabilityInvokeResult(
                status="queued",
                execution_ref="job_opaque",
            )
        }
    )
    result = execute_post_submit_actions(
        accepted,
        FormPostSubmitPlan(
            actions=(automation_workflow_action(mapping={"email": "email"}),)
        ),
        context=_CTX,
        invoker=invoker,
        definition=definition,
    )
    assert result.actions[0].status == "queued"
    assert result.actions[0].execution_ref == "job_opaque"


def test_email_payload_is_not_automation_input() -> None:
    definition = contact_block_compatible_definition()
    accepted = _accept(
        definition,
        {"name": "Ada", "email": "ada@example.com", "message": "Hi"},
    )
    plan = FormPostSubmitPlan(
        actions=(
            email_send_action(),
            automation_workflow_action(),
        )
    )
    invoker = RecordingCapabilityInvoker()
    execute_post_submit_actions(
        accepted, plan, context=_CTX, invoker=invoker, definition=definition
    )
    assert invoker.calls[0]["capability"] == "email.send"
    assert invoker.calls[1]["capability"] == "automation.workflow.trigger"
    assert invoker.calls[1]["payload"]["input"]["email"] == "ada@example.com"
    assert "to" not in invoker.calls[1]["payload"]["input"]