from __future__ import annotations

import json
import subprocess
from urllib.parse import parse_qs

from external_connector.app import create_app
from external_connector.providers.composite import CompositeProvider
from external_connector.providers.fake import FakeProvider
from external_connector.providers.feishu_bot import FeishuBotProvider
from external_connector.providers.weixin_ilink import WeixinIlinkProvider
from external_connector.state import SidecarStateStore
from fastapi.testclient import TestClient


def test_feishu_node_event_utils() -> None:
    result = subprocess.run(
        ["node", "--test", "tests/node/feishu_event_utils.test.js"],
        check=False,
        capture_output=True,
        text=True,
    )

    assert result.returncode == 0, result.stdout + result.stderr


class FakeResponse:
    def __init__(self, payload: dict[str, object]) -> None:
        self.payload = payload
        self.status_code = 200
        self.is_success = True
        self.text = json.dumps(payload)

    def raise_for_status(self) -> None:
        return None

    def json(self) -> dict[str, object]:
        return self.payload


class FakeHttpClient:
    def __init__(self) -> None:
        self.posts: list[tuple[str, dict[str, object] | None, dict[str, str] | None]] = []
        self.registration_poll_response: dict[str, object] = {"error": "authorization_pending"}

    def post(
        self,
        url: str,
        *,
        json: dict[str, object] | None = None,
        data: str | None = None,
        headers: dict[str, str] | None = None,
        timeout: float | None = None,
    ) -> FakeResponse:
        self.posts.append((url, json, headers))
        if url.endswith("/oauth/v1/app/registration"):
            params = parse_qs(data or "")
            action = str((params.get("action") or [""])[0])
            if action == "init":
                return FakeResponse({"supported_auth_methods": ["client_secret"]})
            if action == "begin":
                return FakeResponse(
                    {
                        "verification_uri_complete": "https://accounts.feishu.cn/scan?device=1",
                        "device_code": "device-1",
                        "interval": 1,
                        "expire_in": 600,
                    }
                )
            if action == "poll":
                return FakeResponse(self.registration_poll_response)
        if url.endswith("/open-apis/auth/v3/tenant_access_token/internal"):
            return FakeResponse({"code": 0, "tenant_access_token": "tenant-token", "expire": 7200})
        if "/open-apis/im/v1/messages" in url:
            return FakeResponse({"code": 0, "data": {"message_id": "om_out"}})
        raise AssertionError(url)


def _provider(
    tmp_path,
    *,
    bridge_posts: list[tuple[str, dict[str, object], dict[str, str]]] | None = None,
    http_client: FakeHttpClient | None = None,
    receiver_starts: list[str] | None = None,
) -> FeishuBotProvider:
    def bridge_post(url: str, payload: dict[str, object], headers: dict[str, str]) -> None:
        if bridge_posts is not None:
            bridge_posts.append((url, payload, headers))

    def start_receiver(session) -> object:
        if receiver_starts is not None:
            receiver_starts.append(session.connection_id)
        return object()

    return FeishuBotProvider(
        store=SidecarStateStore(tmp_path / "state.json"),
        http_client=http_client or FakeHttpClient(),
        bridge_base_url="http://beaver:8080",
        public_base_url="http://public-sidecar:8787",
        bridge_token="bridge-token",
        bridge_post=bridge_post,
        receiver_start=start_receiver,
    )


def test_feishu_bot_provider_starts_create_session_with_qr_from_registration(tmp_path) -> None:
    provider = _provider(tmp_path)

    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"mode": "create", "domain": "feishu"},
        }
    )

    assert session["status"] == "qr_ready"
    assert session["qrCode"] == "https://accounts.feishu.cn/scan?device=1&from=onboard"
    assert session["qrImage"].startswith("data:image/svg+xml;base64,")
    assert any("一键创建飞书机器人" in item for item in session["instructions"])
    assert any("/feishu start" in item for item in session["instructions"])
    assert session["metadata"]["eventCallbackPath"] == "/feishu/events"
    assert session["metadata"]["eventCallbackUrl"] == "http://public-sidecar:8787/feishu/events"
    assert session["metadata"]["deviceCode"] == "device-1"


def test_feishu_bot_provider_poll_connects_after_qr_confirmation(tmp_path) -> None:
    http = FakeHttpClient()
    receiver_starts: list[str] = []
    provider = _provider(tmp_path, http_client=http, receiver_starts=receiver_starts)
    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"mode": "create", "domain": "feishu"},
        }
    )
    http.registration_poll_response = {
        "client_id": "cli_qr",
        "client_secret": "qr-secret",
        "user_info": {"tenant_brand": "feishu", "open_id": "ou_me"},
    }

    connected = provider.get_session(session["sessionId"])
    repeated = provider.get_session(session["sessionId"])

    assert connected["status"] == "connected"
    assert repeated["status"] == "connected"
    assert connected["accountId"] == "feishu:cli_qr"
    assert receiver_starts == ["conn_1"]
    stored = provider.store.get_session(session["sessionId"])
    assert stored.metadata["appId"] == "cli_qr"
    assert stored.metadata["appSecret"] == "qr-secret"
    assert stored.metadata["tenantAccessToken"] == "tenant-token"


def test_feishu_bot_provider_connects_with_app_credentials(tmp_path) -> None:
    receiver_starts: list[str] = []
    provider = _provider(tmp_path, receiver_starts=receiver_starts)

    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret", "verificationToken": "verify-token"},
        }
    )

    assert session["status"] == "connected"
    assert session["accountId"] == "feishu:cli_xxx"
    assert session["displayName"] == "Feishu Main"
    assert receiver_starts == ["conn_1"]


def test_feishu_bot_provider_stores_runtime_policy_options(tmp_path) -> None:
    receiver_starts: list[str] = []
    provider = _provider(tmp_path, receiver_starts=receiver_starts)

    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {
                "appId": "cli_xxx",
                "appSecret": "secret",
                "verificationToken": "verify-token",
                "requireMentionInGroups": True,
                "allowFrom": ["ou_1"],
                "groupAllowFrom": ["oc_1"],
                "maxMessageChars": 1234,
            },
        }
    )

    stored = provider.store.get_session(session["sessionId"])
    assert stored.metadata["requireMentionInGroups"] is True
    assert stored.metadata["allowFrom"] == ["ou_1"]
    assert stored.metadata["groupAllowFrom"] == ["oc_1"]
    assert stored.metadata["maxMessageChars"] == 1234


def test_feishu_bot_provider_send_uses_tenant_token_and_dedupes(tmp_path) -> None:
    provider = _provider(tmp_path)
    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret"},
        }
    )
    payload = {
        "requestId": "out_1",
        "connectionId": "conn_1",
        "channelId": "feishu-main",
        "kind": "feishu",
        "target": {"peerId": "ou_user", "peerType": "dm", "threadId": None},
        "content": "hello",
        "metadata": {},
    }

    first = provider.send(payload)
    second = provider.send(payload)

    send_posts = [item for item in provider.http.posts if "/open-apis/im/v1/messages" in item[0]]
    assert session["status"] == "connected"
    assert first == second
    assert first["providerMessageId"] == "om_out"
    assert len(send_posts) == 1
    assert send_posts[0][0].startswith("https://open.feishu.cn/open-apis/im/v1/messages")
    assert send_posts[0][2]["Authorization"] == "Bearer tenant-token"
    assert send_posts[0][1]["receive_id"] == "ou_user"
    assert send_posts[0][1]["msg_type"] == "text"


def test_feishu_bot_provider_send_uses_chat_id_for_group_targets(tmp_path) -> None:
    provider = _provider(tmp_path)
    provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret"},
        }
    )

    result = provider.send(
        {
            "requestId": "out_group_1",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "kind": "feishu",
            "target": {"peerId": "oc_group", "peerType": "group", "threadId": "oc_group"},
            "content": "hello group",
            "metadata": {},
        }
    )

    send_posts = [item for item in provider.http.posts if "/open-apis/im/v1/messages" in item[0]]
    assert result["ok"] is True
    assert send_posts[-1][0].endswith("?receive_id_type=chat_id")
    assert send_posts[-1][1]["receive_id"] == "oc_group"


def test_feishu_bot_provider_send_chunks_oversized_text(tmp_path) -> None:
    provider = _provider(tmp_path)
    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret", "maxMessageChars": 5},
        }
    )

    result = provider.send(
        {
            "requestId": "out_chunked",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "kind": "feishu",
            "target": {"peerId": "ou_user", "peerType": "dm", "threadId": None},
            "content": "helloworld!",
            "metadata": {},
        }
    )

    send_posts = [item for item in provider.http.posts if "/open-apis/im/v1/messages" in item[0]]
    contents = [json.loads(item[1]["content"])["text"] for item in send_posts]
    assert session["status"] == "connected"
    assert result["ok"] is True
    assert contents == ["hello", "world", "!"]


def test_feishu_event_route_requires_known_verification_token_for_challenge(tmp_path) -> None:
    provider = _provider(tmp_path)
    app = create_app(provider=provider, api_token="sidecar-token")

    with TestClient(app) as client:
        response = client.post("/feishu/events", json={"challenge": "abc"})

    assert response.status_code == 401


def test_feishu_event_route_returns_challenge_for_matching_token(tmp_path) -> None:
    provider = _provider(tmp_path)
    provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret", "verificationToken": "verify-token"},
        }
    )
    app = create_app(provider=provider, api_token="sidecar-token")

    with TestClient(app) as client:
        response = client.post("/feishu/events", json={"challenge": "abc", "token": "verify-token"})

    assert response.status_code == 200
    assert response.json() == {"challenge": "abc"}


def test_feishu_event_route_forwards_message_to_bridge(tmp_path) -> None:
    bridge_posts: list[tuple[str, dict[str, object], dict[str, str]]] = []
    provider = _provider(tmp_path, bridge_posts=bridge_posts)
    provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret", "verificationToken": "verify-token"},
        }
    )
    app = create_app(provider=provider, api_token="sidecar-token")

    with TestClient(app) as client:
        response = client.post(
            "/feishu/events",
            json={
                "schema": "2.0",
                "header": {"event_id": "evt_1", "token": "verify-token", "app_id": "cli_xxx"},
                "event": {
                    "sender": {"sender_id": {"open_id": "ou_user"}},
                    "message": {
                        "message_id": "om_1",
                        "chat_id": "oc_chat",
                        "chat_type": "p2p",
                        "message_type": "text",
                        "content": "{\"text\":\"hello feishu\"}",
                    },
                },
            },
        )

    assert response.status_code == 200
    assert response.json() == {"ok": True}
    assert bridge_posts[0][0] == "http://beaver:8080/api/channel-connector-bridge/events"
    assert bridge_posts[0][2]["Authorization"] == "Bearer bridge-token"
    assert bridge_posts[0][1]["eventId"] == "evt_1"
    assert bridge_posts[0][1]["content"] == "hello feishu"
    assert bridge_posts[0][1]["peerId"] == "ou_user"


def test_feishu_event_route_uses_session_callback_base_url(tmp_path) -> None:
    bridge_posts: list[tuple[str, dict[str, object], dict[str, str]]] = []
    provider = _provider(tmp_path, bridge_posts=bridge_posts)
    provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://app-instance-jaychen:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret", "verificationToken": "verify-token"},
        }
    )
    app = create_app(provider=provider, api_token="sidecar-token")

    with TestClient(app) as client:
        response = client.post(
            "/feishu/events",
            json={
                "schema": "2.0",
                "header": {"event_id": "evt_1", "token": "verify-token", "app_id": "cli_xxx"},
                "event": {
                    "sender": {"sender_id": {"open_id": "ou_user"}},
                    "message": {
                        "message_id": "om_1",
                        "chat_id": "oc_chat",
                        "chat_type": "p2p",
                        "message_type": "text",
                        "content": "{\"text\":\"hello feishu\"}",
                    },
                },
            },
        )

    assert response.status_code == 200
    assert bridge_posts[0][0] == "http://app-instance-jaychen:8080/api/channel-connector-bridge/events"


def test_feishu_event_route_ignores_bot_sender_and_platform_commands(tmp_path) -> None:
    bridge_posts: list[tuple[str, dict[str, object], dict[str, str]]] = []
    provider = _provider(tmp_path, bridge_posts=bridge_posts)
    provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"appId": "cli_xxx", "appSecret": "secret", "verificationToken": "verify-token"},
        }
    )
    app = create_app(provider=provider, api_token="sidecar-token")

    with TestClient(app) as client:
        bot = client.post(
            "/feishu/events",
            json={
                "header": {"event_id": "evt_bot", "token": "verify-token", "app_id": "cli_xxx"},
                "event": {
                    "sender": {"sender_type": "bot", "sender_id": {"open_id": "ou_bot"}},
                    "message": {
                        "message_id": "om_bot",
                        "chat_id": "oc_chat",
                        "chat_type": "p2p",
                        "message_type": "text",
                        "content": "{\"text\":\"hello\"}",
                    },
                },
            },
        )
        command = client.post(
            "/feishu/events",
            json={
                "header": {"event_id": "evt_command", "token": "verify-token", "app_id": "cli_xxx"},
                "event": {
                    "sender": {"sender_type": "user", "sender_id": {"open_id": "ou_user"}},
                    "message": {
                        "message_id": "om_command",
                        "chat_id": "oc_chat",
                        "chat_type": "p2p",
                        "message_type": "text",
                        "content": "{\"text\":\"/feishu start\"}",
                    },
                },
            },
        )

    assert bot.status_code == 200
    assert command.status_code == 200
    assert bot.json() == {"ok": True, "ignored": "sender_type:bot"}
    assert command.json() == {"ok": True, "ignored": "feishu_command"}
    assert bridge_posts == []


def test_composite_provider_routes_feishu_and_weixin_descriptors(tmp_path) -> None:
    store = SidecarStateStore(tmp_path / "state.json")
    provider = CompositeProvider([FakeProvider(store), _provider(tmp_path)])

    connectors = provider.connectors()

    assert [item["kind"] for item in connectors] == ["weixin", "feishu", "feishu"]
    assert provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {},
        }
    )["status"] == "qr_ready"


def test_composite_provider_get_session_routes_feishu_session_to_feishu_provider(tmp_path) -> None:
    http = FakeHttpClient()
    store = SidecarStateStore(tmp_path / "state.json")
    provider = CompositeProvider(
        [
            WeixinIlinkProvider(
                store=store,
                http_client=FakeHttpClient(),
                bridge_base_url="http://beaver:8080",
                bridge_token="bridge-token",
                start_receivers=False,
            ),
            FeishuBotProvider(
                store=store,
                http_client=http,
                bridge_base_url="http://beaver:8080",
                public_base_url="http://public-sidecar:8787",
                bridge_token="bridge-token",
                start_receivers=False,
            ),
        ]
    )
    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {"mode": "create", "domain": "feishu"},
        }
    )
    http.registration_poll_response = {
        "client_id": "cli_qr",
        "client_secret": "qr-secret",
        "user_info": {"tenant_brand": "feishu", "open_id": "ou_me"},
    }

    connected = provider.get_session(session["sessionId"])

    assert connected["status"] == "connected"
    assert connected["accountId"] == "feishu:cli_qr"