import os
import tempfile
import threading
import unittest
from types import SimpleNamespace
from unittest.mock import patch

from modules.quantri import lead

try:
    import fcntl
except ImportError:  # pragma: no cover - production hosting is Unix-based.
    fcntl = None


class FakeClient:
    def __init__(self):
        self.ADMIN = ["100"]
        self.ADM = []
        self.uid = "999"
        self.command_handler = SimpleNamespace(current_prefix=".")
        self.replies = []
        self.groups = {}
        self.phone_results = {}
        self.friend_requests = []
        self.messages = []
        self.on_send = None

    def replyMessage(self, message, reply_target, thread_id, thread_type, ttl=0):
        self.replies.append(message.text)

    def fetchGroupInfo(self, group_id):
        group_id = str(group_id)
        return SimpleNamespace(gridInfoMap={group_id: self.groups[group_id]})

    def fetchPhoneNumber(self, phone, language="en"):
        result = self.phone_results[phone]
        if isinstance(result, BaseException):
            raise result
        return result

    def sendFriendRequest(self, user_id, message):
        self.friend_requests.append((str(user_id), message))
        if self.on_send:
            self.on_send()
        return SimpleNamespace(msgId=f"friend-{user_id}")

    def send(self, message, thread_id, thread_type):
        self.messages.append((str(thread_id), message.text, thread_type))
        if self.on_send:
            self.on_send()
        return SimpleNamespace(msgId=f"message-{thread_id}")


class LeadCommandTests(unittest.TestCase):
    def setUp(self):
        self.temp_dir = tempfile.TemporaryDirectory()
        self.state_file = os.path.join(self.temp_dir.name, "lead_state.json")
        self.state_patch = patch.object(lead, "STATE_FILE", self.state_file)
        self.state_patch.start()
        lead._active_runs.clear()

    def tearDown(self):
        for campaign_id in list(lead._run_lock_handles):
            lead._release_run(campaign_id)
        lead._active_runs.clear()
        self.state_patch.stop()
        self.temp_dir.cleanup()

    def test_phone_import_requires_consent_and_deduplicates(self):
        state = lead.default_state()

        report = lead.import_phone_lines(
            state,
            "\n".join(
                [
                    "090 123 4567 | Nguyen Van A | form website",
                    "+84 901234567 | Ban trung | landing page",
                    "0912345678 | Thieu nguon consent",
                    "khong-phai-so | Loi | form website",
                ]
            ),
        )

        self.assertEqual(report, {"added": 1, "duplicates": 1, "rejected": 2})
        self.assertEqual(list(state["leads"]), ["phone:84901234567"])
        self.assertEqual(
            state["leads"]["phone:84901234567"]["consent_source"],
            "form website",
        )

    def test_group_import_uses_only_visible_members(self):
        client = FakeClient()
        client.groups["group-1"] = {
            "setting": {"lockViewMember": 0},
            "memVerList": ["101_An", "102_Binh_Nguyen", "bad", "101_Trung"],
        }
        state = lead.default_state()

        report = lead.import_group_members(
            state, client, "group-1", "dang ky tai su kien"
        )

        self.assertEqual(report, {"added": 2, "duplicates": 1, "rejected": 1})
        self.assertEqual(set(state["leads"]), {"uid:101", "uid:102"})
        self.assertEqual(state["leads"]["uid:102"]["name"], "Binh_Nguyen")

        client.groups["group-2"] = {
            "setting": {"lockViewMember": 1},
            "memVerList": ["103_Cuong"],
        }
        with self.assertRaisesRegex(ValueError, "ẩn danh sách thành viên"):
            lead.import_group_members(
                state, client, "group-2", "dang ky tai su kien"
            )

        client.groups["group-3"] = {
            "setting": {"lockViewMember": "0"},
            "memVerList": ["103_Cuong"],
        }
        report = lead.import_group_members(
            state, client, "group-3", "dang ky tai su kien"
        )
        self.assertEqual(report["added"], 1)

    def test_command_is_admin_only_and_normalizes_admin_ids(self):
        client = FakeClient()
        message_object = SimpleNamespace(quote=None)

        with patch.object(lead, "ADMIN", ["100"]):
            lead.handle_lead_command(
                ".lead add 0901234567 | A | form",
                message_object,
                "control-group",
                "GROUP",
                "not-admin",
                client,
            )
            self.assertIn("không có quyền", client.replies[-1].casefold())
            self.assertFalse(os.path.exists(self.state_file))

            lead.handle_lead_command(
                ".lead add 0901234567 | A | form",
                message_object,
                "control-group",
                "USER",
                100,
                client,
            )

        self.assertIn("đã thêm", client.replies[-1].casefold())
        self.assertIn("phone:84901234567", lead.load_state()["leads"])

    def test_blocklist_excludes_lead_from_new_campaign(self):
        state = lead.default_state()
        lead.import_phone_lines(state, "0901234567 | A | form")
        lead.block_target(state, "0901234567", reason="opt-out")

        with self.assertRaisesRegex(ValueError, "(?i)không có lead hợp lệ"):
            lead.create_campaign(state, "100", "message", "Xin chao")

    def test_lookup_uid_uses_local_phone_and_rejects_api_error_objects(self):
        client = FakeClient()
        client.phone_results["0901234567"] = SimpleNamespace(
            uid=123, zalo_name="Nguyen A"
        )

        uid, name = lead.lookup_uid(client, "84901234567")

        self.assertEqual((uid, name), ("123", "Nguyen A"))

        client.phone_results["0901234568"] = SimpleNamespace(
            error_code=1337, error_message="hidden"
        )
        with self.assertRaisesRegex(ValueError, "Không tìm thấy UID"):
            lead.lookup_uid(client, "84901234568")

    def test_friend_campaign_resolves_phone_and_sends_each_uid_once(self):
        client = FakeClient()
        client.phone_results["0901234567"] = SimpleNamespace(
            uid="101", zalo_name="An"
        )
        state = lead.default_state()
        lead.import_phone_lines(state, "0901234567 | An | form")
        state["leads"]["uid:202"] = lead._lead_record(
            uid="202", name="Binh", consent_source="event", source_type="group"
        )
        campaign = lead.create_campaign(state, "100", "friend", "Xin chao")
        lead.save_state(state)

        report = lead.send_campaign(
            client, campaign["id"], sleep_fn=lambda _seconds: None
        )

        self.assertEqual(
            client.friend_requests, [("101", "Xin chao"), ("202", "Xin chao")]
        )
        self.assertEqual(report["sent_count"], 2)
        saved = lead.load_state()["campaigns"][campaign["id"]]
        self.assertEqual(saved["status"], "completed")
        self.assertEqual(len(saved["sent"]), 2)

    def test_message_campaign_uses_user_thread(self):
        client = FakeClient()
        state = lead.default_state()
        state["leads"]["uid:101"] = lead._lead_record(
            uid="101", name="An", consent_source="form", source_type="group"
        )
        campaign = lead.create_campaign(state, "100", "message", "Noi dung")
        lead.save_state(state)

        lead.send_campaign(client, campaign["id"], sleep_fn=lambda _seconds: None)

        self.assertEqual(
            client.messages, [("101", "Noi dung", lead.USER_THREAD_TYPE)]
        )

    def test_confirmation_is_admin_bound_and_single_use(self):
        state = lead.default_state()
        state["leads"]["uid:101"] = lead._lead_record(
            uid="101", consent_source="form", source_type="group"
        )
        campaign = lead.create_campaign(state, "100", "friend", "Xin chao")
        lead.save_state(state)
        token = lead.issue_confirmation(campaign["id"], "send")

        with self.assertRaisesRegex(ValueError, "người tạo"):
            lead.consume_confirmation(campaign["id"], token, "200")
        action = lead.consume_confirmation(campaign["id"], token, "100")
        self.assertEqual(action, "send")
        with self.assertRaisesRegex(ValueError, "không đúng|đã được sử dụng"):
            lead.consume_confirmation(campaign["id"], token, "100")

    def test_stop_halts_before_next_lead(self):
        client = FakeClient()
        state = lead.default_state()
        for uid in ("101", "102"):
            state["leads"][f"uid:{uid}"] = lead._lead_record(
                uid=uid, consent_source="form", source_type="group"
            )
        campaign = lead.create_campaign(state, "100", "friend", "Xin chao")
        lead.save_state(state)

        def request_stop():
            current = lead.load_state()
            current["campaigns"][campaign["id"]]["cancel_requested"] = True
            lead.save_state(current)
            client.on_send = None

        client.on_send = request_stop
        report = lead.send_campaign(
            client, campaign["id"], sleep_fn=lambda _seconds: None
        )

        self.assertEqual(client.friend_requests, [("101", "Xin chao")])
        self.assertEqual(report["pending_count"], 1)
        self.assertEqual(report["status"], "cancelled")

    def test_retry_targets_only_retryable_pre_send_failures(self):
        client = FakeClient()
        state = lead.default_state()
        for uid in ("101", "102", "103"):
            state["leads"][f"uid:{uid}"] = lead._lead_record(
                uid=uid, consent_source="form", source_type="group"
            )
        campaign = lead.create_campaign(state, "100", "friend", "Xin chao")
        campaign["sent"]["uid:101"] = {"uid": "101", "sent_at": lead._now_iso()}
        campaign["failed"]["uid:102"] = {
            "reason": "lookup failed",
            "retryable": True,
        }
        campaign["uncertain"]["uid:103"] = {
            "uid": "103",
            "reason": "unknown delivery",
        }
        lead.save_state(state)

        lead.send_campaign(
            client,
            campaign["id"],
            retry_failed=True,
            sleep_fn=lambda _seconds: None,
        )

        self.assertEqual(client.friend_requests, [("102", "Xin chao")])

    def test_import_command_reads_quoted_multiline_with_default_consent(self):
        client = FakeClient()
        message_object = SimpleNamespace(
            quote=SimpleNamespace(content="0901234567 | An\n0912345678 | Binh")
        )

        with patch.object(lead, "ADMIN", ["100"]):
            lead.handle_lead_command(
                ".lead import form webinar",
                message_object,
                "control-group",
                "USER",
                "100",
                client,
            )

        state = lead.load_state()
        self.assertEqual(len(state["leads"]), 2)
        self.assertTrue(
            all(item["consent_source"] == "form webinar" for item in state["leads"].values())
        )

    def test_command_creates_campaign_sets_delay_and_issues_confirmation(self):
        client = FakeClient()
        client.command_handler.current_prefix = "!"
        message_object = SimpleNamespace(quote=None)
        state = lead.default_state()
        state["leads"]["uid:101"] = lead._lead_record(
            uid="101", consent_source="form", source_type="group"
        )
        lead.save_state(state)

        with patch.object(lead, "ADMIN", ["100"]):
            lead.handle_lead_command(
                "!lead nhantin Xin chao khach hang",
                message_object,
                "control-group",
                "USER",
                "100",
                client,
            )
            lead.handle_lead_command(
                "!lead nghi 2",
                message_object,
                "control-group",
                "USER",
                "100",
                client,
            )
            lead.handle_lead_command(
                "!lead gui",
                message_object,
                "control-group",
                "USER",
                "100",
                client,
            )

        saved = lead.load_state()
        campaign_id = saved["active_by_admin"]["100"]
        campaign = saved["campaigns"][campaign_id]
        self.assertEqual(campaign["action"], "message")
        self.assertEqual(campaign["delay_seconds"], 2)
        self.assertEqual(campaign["status"], "awaiting_confirmation")
        self.assertRegex(client.replies[-1], r"!lead xacnhan [A-F0-9]{6}")

    def test_confirm_command_runs_campaign_and_releases_run_lock(self):
        client = FakeClient()
        message_object = SimpleNamespace(quote=None)
        state = lead.default_state()
        state["leads"]["uid:101"] = lead._lead_record(
            uid="101", consent_source="form", source_type="group"
        )
        campaign = lead.create_campaign(state, "100", "message", "Xin chao")
        lead.save_state(state)
        token = lead.issue_confirmation(campaign["id"], "send")

        class ImmediateThread:
            def __init__(self, target, args, daemon):
                self.target = target
                self.args = args

            def start(self):
                self.target(*self.args)

        with patch.object(lead, "ADMIN", ["100"]), patch.object(
            lead.threading, "Thread", ImmediateThread
        ):
            lead.handle_lead_command(
                f".lead xacnhan {token}",
                message_object,
                "control-group",
                "USER",
                "100",
                client,
            )

        self.assertEqual(client.messages, [("101", "Xin chao", lead.USER_THREAD_TYPE)])
        self.assertNotIn(campaign["id"], lead._active_runs)
        self.assertIn("hoàn tất", " ".join(client.replies).casefold())

    def test_send_error_stays_uncertain_and_is_not_retryable(self):
        client = FakeClient()
        state = lead.default_state()
        state["leads"]["uid:101"] = lead._lead_record(
            uid="101", consent_source="form", source_type="group"
        )
        campaign = lead.create_campaign(state, "100", "friend", "Xin chao")
        lead.save_state(state)

        def fail_send():
            raise RuntimeError("timeout")

        client.on_send = fail_send
        report = lead.send_campaign(
            client, campaign["id"], sleep_fn=lambda _seconds: None
        )

        self.assertEqual(report["uncertain_count"], 1)
        saved = lead.load_state()["campaigns"][campaign["id"]]
        self.assertEqual(lead.build_pending_targets(saved, retry_failed=True), [])

    def test_empty_api_response_stays_uncertain(self):
        client = FakeClient()
        client.sendFriendRequest = lambda _uid, _message: None
        state = lead.default_state()
        state["leads"]["uid:101"] = lead._lead_record(
            uid="101", consent_source="form", source_type="group"
        )
        campaign = lead.create_campaign(state, "100", "friend", "Xin chao")
        lead.save_state(state)

        report = lead.send_campaign(
            client, campaign["id"], sleep_fn=lambda _seconds: None
        )

        self.assertEqual(report["sent_count"], 0)
        self.assertEqual(report["uncertain_count"], 1)

    def test_daily_limit_pauses_before_calling_zalo(self):
        client = FakeClient()
        state = lead.default_state()
        today = lead._now_iso()
        for index in range(lead.DAILY_LIMITS["friend"]):
            state["delivery_ledger"][f"friend:old-{index}"] = {
                "action": "friend",
                "status": "sent",
                "attempted_at": today,
            }
        state["leads"]["uid:101"] = lead._lead_record(
            uid="101", consent_source="form", source_type="group"
        )
        campaign = lead.create_campaign(state, "100", "friend", "Xin chao")
        lead.save_state(state)

        report = lead.send_campaign(
            client, campaign["id"], sleep_fn=lambda _seconds: None
        )

        self.assertEqual(client.friend_requests, [])
        self.assertEqual(report["status"], "paused_limit")
        self.assertEqual(report["pending_count"], 1)

    def test_sensitive_commands_require_private_chat(self):
        client = FakeClient()

        with patch.object(lead, "ADMIN", ["100"]):
            lead.handle_lead_command(
                ".lead ds",
                SimpleNamespace(quote=None),
                "control-group",
                "GROUP",
                "100",
                client,
            )

        self.assertIn("chat riêng", client.replies[-1].casefold())

    @unittest.skipIf(fcntl is None, "fcntl is required for cross-process locking")
    def test_external_state_lock_serializes_reads(self):
        lead.save_state(lead.default_state())
        lock_path = f"{self.state_file}.lock"
        reader_started = threading.Event()
        reader_finished = threading.Event()

        def read_state():
            reader_started.set()
            lead.load_state()
            reader_finished.set()

        with open(lock_path, "a+", encoding="utf-8") as external_lock:
            fcntl.flock(external_lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            reader = threading.Thread(target=read_state)
            reader.start()
            self.assertTrue(reader_started.wait(timeout=1))
            self.assertFalse(reader_finished.wait(timeout=0.05))
            fcntl.flock(external_lock.fileno(), fcntl.LOCK_UN)
            self.assertTrue(reader_finished.wait(timeout=1))
            reader.join(timeout=1)


if __name__ == "__main__":
    unittest.main()