from __future__ import annotations

import json
import socket
from datetime import datetime, timedelta, timezone
from typing import Any, Callable

from sqlalchemy import select

from services.sales_service import sales_events as sales_event_types
from services.sales_service.deal_state_machine import DealStateMachineError, transition_deal_stage
from services.sales_service.event_publisher import SalesEventPublisher
from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.sales_sql_models import (
    SalesAutomationTaskRow,
    SalesChannelSwitchRow,
    SalesCommunicationSessionRow,
    SalesDealRow,
    SalesEscalationRow,
    SalesInvoiceRow,
    SalesPipelineStageRow,
)


RETRY_DELAYS_SECONDS = {
    1: 60,
    2: 300,
    3: 900,
}
ACTIVE_TASK_STATUSES = {"pending", "running"}
TERMINAL_TASK_STATUSES = {"completed", "failed", "canceled"}


def _now_dt() -> datetime:
    return datetime.now(timezone.utc).replace(microsecond=0)


def _iso(dt: datetime) -> str:
    return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat()


def _payload(row: SalesAutomationTaskRow) -> dict[str, Any]:
    try:
        parsed = json.loads(row.payload_json or "{}")
    except json.JSONDecodeError:
        return {}
    return parsed if isinstance(parsed, dict) else {}


def _set_payload(row: SalesAutomationTaskRow, payload: dict[str, Any]) -> None:
    row.payload_json = json.dumps(payload, ensure_ascii=False)


def _task_error_message(exc: Exception) -> str:
    message = str(exc).strip()
    return message[:1000] if message else exc.__class__.__name__


class SalesAutomationWorker:
    def __init__(self, *, worker_id: str | None = None, batch_size: int = 10) -> None:
        self.worker_id = worker_id or f"sales-worker-{socket.gethostname()}-{new_id('wrk')}"
        self.batch_size = batch_size
        self.handlers: dict[str, Callable[[Any, SalesAutomationTaskRow], None]] = {
            "follow_up_customer": self._handle_follow_up_customer,
            "mark_invoice_overdue": self._handle_mark_invoice_overdue,
            "send_invoice_reminder": self._handle_send_invoice_reminder,
            "recommend_channel_switch": self._handle_recommend_channel_switch,
            "close_stale_communication": self._handle_close_stale_communication,
            "escalate_to_human": self._handle_escalate_to_human,
            "post_sale_transfer": self._handle_post_sale_transfer,
        }

    def claim_pending_tasks(self, session, *, limit: int | None = None, now: str | None = None) -> list[SalesAutomationTaskRow]:
        claim_limit = limit or self.batch_size
        now_text = now or utc_now_iso()
        stmt = (
            select(SalesAutomationTaskRow)
            .where(
                SalesAutomationTaskRow.status == "pending",
                SalesAutomationTaskRow.run_at <= now_text,
            )
            .order_by(SalesAutomationTaskRow.run_at.asc(), SalesAutomationTaskRow.id.asc())
            .limit(claim_limit)
        )
        if session.bind is not None and session.bind.dialect.name == "postgresql":
            stmt = stmt.with_for_update(skip_locked=True)
        rows = session.execute(stmt).scalars().all()
        for row in rows:
            row.status = "running"
            row.locked_at = now_text
            row.locked_by = self.worker_id
            row.last_error = None
            row.updated_at = now_text
        session.flush()
        return list(rows)

    def run_once(self, *, limit: int | None = None) -> int:
        session = get_session()
        try:
            tasks = self.claim_pending_tasks(session, limit=limit)
            task_ids = [task.task_id for task in tasks]
            session.commit()
        finally:
            session.close()

        for task_id in task_ids:
            self._execute_claimed_task(task_id)
        return len(task_ids)

    def _execute_claimed_task(self, task_id: str) -> None:
        session = get_session()
        try:
            task = session.execute(
                select(SalesAutomationTaskRow).where(
                    SalesAutomationTaskRow.task_id == task_id,
                    SalesAutomationTaskRow.status == "running",
                    SalesAutomationTaskRow.locked_by == self.worker_id,
                )
            ).scalar_one_or_none()
            if task is None:
                session.rollback()
                return
            try:
                self._execute_task(session, task)
            except Exception as exc:  # noqa: BLE001
                self._mark_failed_or_retry(task, exc)
            else:
                self._mark_completed(task)
            session.commit()
        finally:
            session.close()

    def _execute_task(self, session, task: SalesAutomationTaskRow) -> None:
        handler = self.handlers.get(task.task_type)
        if handler is None:
            raise RuntimeError(f"Unsupported automation task type: {task.task_type}")
        handler(session, task)

    def _mark_completed(self, task: SalesAutomationTaskRow) -> None:
        now = utc_now_iso()
        task.status = "completed"
        task.completed_at = now
        task.failed_at = None
        task.locked_at = None
        task.locked_by = None
        task.last_error = None
        task.updated_at = now

    def _mark_failed_or_retry(self, task: SalesAutomationTaskRow, exc: Exception) -> None:
        now_dt = _now_dt()
        task.retry_count += 1
        task.last_error = _task_error_message(exc)
        task.locked_at = None
        task.locked_by = None
        task.updated_at = _iso(now_dt)
        max_retries = int(task.max_retries or 3)
        if task.retry_count < max_retries:
            task.status = "pending"
            delay_seconds = RETRY_DELAYS_SECONDS.get(task.retry_count, 900)
            task.run_at = _iso(now_dt + timedelta(seconds=delay_seconds))
            return
        task.status = "failed"
        task.failed_at = _iso(now_dt)

    def _get_deal(self, session, task: SalesAutomationTaskRow) -> SalesDealRow:
        row = session.execute(
            select(SalesDealRow).where(
                SalesDealRow.tenant_id == task.tenant_id,
                SalesDealRow.deal_id == task.deal_id,
            )
        ).scalar_one_or_none()
        if row is None:
            raise RuntimeError(f"Deal not found: {task.deal_id}")
        return row

    def _get_invoice(self, session, task: SalesAutomationTaskRow, invoice_id: str) -> SalesInvoiceRow:
        row = session.execute(
            select(SalesInvoiceRow).where(
                SalesInvoiceRow.tenant_id == task.tenant_id,
                SalesInvoiceRow.invoice_id == invoice_id,
            )
        ).scalar_one_or_none()
        if row is None:
            raise RuntimeError(f"Invoice not found: {invoice_id}")
        if row.deal_id != task.deal_id:
            raise RuntimeError("Invoice does not belong to task deal")
        return row

    def _stage_code(self, session, deal: SalesDealRow) -> str | None:
        stage = session.execute(
            select(SalesPipelineStageRow).where(
                SalesPipelineStageRow.tenant_id == deal.tenant_id,
                SalesPipelineStageRow.stage_id == deal.stage_id,
            )
        ).scalar_one_or_none()
        return stage.code if stage is not None else None

    def _transition_stage(
        self,
        session,
        *,
        task: SalesAutomationTaskRow,
        target_stage_code: str,
        reason: str,
        force_on_invalid: bool = True,
    ) -> None:
        metadata = {"automation_task_id": task.task_id, "task_type": task.task_type}
        try:
            transition_deal_stage(
                session,
                tenant_id=task.tenant_id,
                deal_id=task.deal_id,
                target_stage_code=target_stage_code,
                actor_type="system",
                reason=reason,
                metadata=metadata,
            )
        except DealStateMachineError as exc:
            if not force_on_invalid or exc.error != "invalid_stage_transition":
                raise
            transition_deal_stage(
                session,
                tenant_id=task.tenant_id,
                deal_id=task.deal_id,
                target_stage_code=target_stage_code,
                actor_type="system",
                reason=reason,
                metadata={**metadata, "force_reason": reason},
                force=True,
            )

    def _publish(
        self,
        session,
        *,
        task: SalesAutomationTaskRow,
        event_type: str,
        aggregate_type: str,
        aggregate_id: str,
        payload: dict[str, Any],
    ) -> None:
        SalesEventPublisher.publish_sales_event(
            session,
            tenant_id=task.tenant_id,
            event_type=event_type,
            aggregate_type=aggregate_type,
            aggregate_id=aggregate_id,
            actor_type="system",
            actor_id=self.worker_id,
            payload={
                "deal_id": task.deal_id,
                "automation_task_id": task.task_id,
                "task_type": task.task_type,
                **payload,
            },
        )

    def _find_related_task(
        self,
        session,
        *,
        tenant_id: str,
        deal_id: str,
        task_type: str,
        match_payload: dict[str, Any],
    ) -> SalesAutomationTaskRow | None:
        rows = session.execute(
            select(SalesAutomationTaskRow)
            .where(
                SalesAutomationTaskRow.tenant_id == tenant_id,
                SalesAutomationTaskRow.deal_id == deal_id,
                SalesAutomationTaskRow.task_type == task_type,
                SalesAutomationTaskRow.status.in_(["pending", "running", "completed"]),
            )
            .order_by(SalesAutomationTaskRow.id.desc())
        ).scalars().all()
        for row in rows:
            payload = _payload(row)
            if all(payload.get(key) == value for key, value in match_payload.items()):
                return row
        return None

    def _create_task_once(
        self,
        session,
        *,
        tenant_id: str,
        deal_id: str,
        task_type: str,
        run_at: str | None = None,
        payload: dict[str, Any] | None = None,
        match_payload: dict[str, Any] | None = None,
    ) -> SalesAutomationTaskRow:
        payload_data = payload or {}
        existing = self._find_related_task(
            session,
            tenant_id=tenant_id,
            deal_id=deal_id,
            task_type=task_type,
            match_payload=match_payload or payload_data,
        )
        if existing is not None:
            return existing
        now = utc_now_iso()
        row = SalesAutomationTaskRow(
            task_id=new_id("tsk"),
            tenant_id=tenant_id,
            deal_id=deal_id,
            task_type=task_type,
            payload_json=json.dumps(payload_data, ensure_ascii=False),
            run_at=run_at or now,
            status="pending",
            retry_count=0,
            max_retries=3,
            locked_at=None,
            locked_by=None,
            completed_at=None,
            failed_at=None,
            last_error=None,
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        return row

    def _handle_follow_up_customer(self, session, task: SalesAutomationTaskRow) -> None:
        deal = self._get_deal(session, task)
        data = _payload(task)
        if deal.status != "active":
            data["automation_result"] = {"status": "skipped", "reason": f"deal_status_{deal.status}"}
            _set_payload(task, data)
            return

        now = utc_now_iso()
        deal.next_action_type = str(data.get("next_action_type") or "follow_up_customer")
        deal.next_action_at = now
        deal.updated_at = now
        current_stage = self._stage_code(session, deal)
        preferred_channel = str(data.get("preferred_channel") or deal.current_channel or deal.preferred_channel or "text")
        if current_stage == "follow_up_scheduled":
            target = "active_voice_communication" if preferred_channel == "voice" else "active_text_communication"
            self._transition_stage(
                session,
                task=task,
                target_stage_code=target,
                reason="automation.follow_up_customer",
            )
        data["automation_result"] = {
            "status": "follow_up_ready",
            "next_action_type": deal.next_action_type,
            "next_action_at": deal.next_action_at,
        }
        _set_payload(task, data)
        self._publish(
            session,
            task=task,
            event_type=sales_event_types.DEAL_FOLLOW_UP_READY,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            payload={"next_action_type": deal.next_action_type, "next_action_at": deal.next_action_at},
        )

    def _handle_mark_invoice_overdue(self, session, task: SalesAutomationTaskRow) -> None:
        data = _payload(task)
        invoice_id = str(data.get("invoice_id") or "").strip()
        if not invoice_id:
            raise RuntimeError("invoice_id is required")
        invoice = self._get_invoice(session, task, invoice_id)
        deal = self._get_deal(session, task)
        if invoice.status in {"paid", "canceled"}:
            data["automation_result"] = {"status": "skipped", "invoice_status": invoice.status}
            _set_payload(task, data)
            return

        previous_status = invoice.status
        now = utc_now_iso()
        invoice.status = "overdue"
        invoice.updated_at = now
        if deal.status == "active":
            self._transition_stage(
                session,
                task=task,
                target_stage_code="payment_overdue",
                reason="automation.mark_invoice_overdue",
            )
        if previous_status != "overdue":
            self._publish(
                session,
                task=task,
                event_type=sales_event_types.INVOICE_OVERDUE,
                aggregate_type="invoice",
                aggregate_id=invoice.invoice_id,
                payload={
                    "invoice_id": invoice.invoice_id,
                    "invoice_number": invoice.invoice_number,
                    "amount": invoice.amount,
                    "currency": invoice.currency,
                    "due_date": invoice.due_date,
                },
            )
        reminder = self._create_task_once(
            session,
            tenant_id=task.tenant_id,
            deal_id=task.deal_id,
            task_type="send_invoice_reminder",
            payload={"invoice_id": invoice.invoice_id, "source_task_id": task.task_id, "reason": "invoice.overdue"},
            match_payload={"invoice_id": invoice.invoice_id, "reason": "invoice.overdue"},
        )
        data["automation_result"] = {"status": "invoice_overdue", "reminder_task_id": reminder.task_id}
        _set_payload(task, data)

    def _handle_send_invoice_reminder(self, session, task: SalesAutomationTaskRow) -> None:
        data = _payload(task)
        invoice_id = str(data.get("invoice_id") or "").strip()
        if not invoice_id:
            raise RuntimeError("invoice_id is required")
        invoice = self._get_invoice(session, task, invoice_id)
        deal = self._get_deal(session, task)
        if invoice.status in {"paid", "canceled"} or deal.status != "active":
            data["automation_result"] = {
                "status": "skipped",
                "invoice_status": invoice.status,
                "deal_status": deal.status,
            }
            _set_payload(task, data)
            return

        now = utc_now_iso()
        deal.next_action_type = "invoice_reminder"
        deal.next_action_at = now
        deal.updated_at = now
        follow_up = self._create_task_once(
            session,
            tenant_id=task.tenant_id,
            deal_id=task.deal_id,
            task_type="follow_up_customer",
            payload={"invoice_id": invoice.invoice_id, "source_task_id": task.task_id, "reason": "invoice.reminder"},
            match_payload={"invoice_id": invoice.invoice_id, "reason": "invoice.reminder"},
        )
        data["automation_result"] = {
            "status": "invoice_reminder_ready",
            "follow_up_task_id": follow_up.task_id,
        }
        _set_payload(task, data)
        self._publish(
            session,
            task=task,
            event_type=sales_event_types.INVOICE_REMINDER_SENT,
            aggregate_type="invoice",
            aggregate_id=invoice.invoice_id,
            payload={
                "invoice_id": invoice.invoice_id,
                "invoice_number": invoice.invoice_number,
                "invoice_status": invoice.status,
                "follow_up_task_id": follow_up.task_id,
            },
        )

    def _handle_recommend_channel_switch(self, session, task: SalesAutomationTaskRow) -> None:
        deal = self._get_deal(session, task)
        data = _payload(task)
        recommendation = {
            "recommended_from_channel": data.get("recommended_from_channel") or deal.current_channel,
            "recommended_to_channel": data.get("recommended_to_channel") or data.get("to_channel") or "voice",
            "reason": data.get("reason") or "automation_recommendation",
        }
        deal.next_action_type = "channel_switch_recommended"
        deal.next_action_at = utc_now_iso()
        deal.updated_at = deal.next_action_at
        data["automation_result"] = recommendation
        if data.get("auto_apply"):
            existing_switch = session.execute(
                select(SalesChannelSwitchRow)
                .where(
                    SalesChannelSwitchRow.tenant_id == task.tenant_id,
                    SalesChannelSwitchRow.deal_id == task.deal_id,
                    SalesChannelSwitchRow.recommended_by_task_id == task.task_id,
                )
                .order_by(SalesChannelSwitchRow.id.desc())
            ).scalars().first()
            if existing_switch is None:
                now = utc_now_iso()
                to_channel = "voice" if recommendation["recommended_to_channel"] == "voice" else "text"
                from_channel = "voice" if recommendation["recommended_from_channel"] == "voice" else "text"
                existing_switch = SalesChannelSwitchRow(
                    switch_id=new_id("swc"),
                    tenant_id=task.tenant_id,
                    deal_id=task.deal_id,
                    communication_id=None,
                    communication_session_id=None,
                    previous_communication_session_id=None,
                    new_communication_session_id=None,
                    from_channel=from_channel,
                    to_channel=to_channel,
                    reason_code=str(data.get("reason_code") or "system_recommendation"),
                    reason_text=recommendation["reason"],
                    reason_for_channel_switch=recommendation["reason"],
                    initiated_by_type="system",
                    initiated_by_id=None,
                    source_type="automation_task",
                    source_id=task.task_id,
                    recommended_by_task_id=task.task_id,
                    switched_at=now,
                    created_at=now,
                )
                session.add(existing_switch)
                deal.current_channel = to_channel
                deal.updated_at = now
                current_stage = self._stage_code(session, deal)
                target_stage = "active_voice_communication" if to_channel == "voice" else "active_text_communication"
                if current_stage in {"active_text_communication", "active_voice_communication"} and current_stage != target_stage:
                    transition_deal_stage(
                        session,
                        tenant_id=task.tenant_id,
                        deal_id=task.deal_id,
                        target_stage_code=target_stage,
                        actor_type="system",
                        reason="automation.recommend_channel_switch",
                        metadata={"automation_task_id": task.task_id, "switch_id": existing_switch.switch_id},
                    )
                self._publish(
                    session,
                    task=task,
                    event_type=sales_event_types.COMMUNICATION_CHANNEL_SWITCHED,
                    aggregate_type="deal",
                    aggregate_id=deal.deal_id,
                    payload={
                        "switch_id": existing_switch.switch_id,
                        "from_channel": existing_switch.from_channel,
                        "to_channel": existing_switch.to_channel,
                        "reason_code": existing_switch.reason_code,
                        "recommended_by_task_id": task.task_id,
                    },
                )
            data["actual_switch_id"] = existing_switch.switch_id
        _set_payload(task, data)
        self._publish(
            session,
            task=task,
            event_type=sales_event_types.DEAL_CHANNEL_SWITCH_RECOMMENDED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            payload=recommendation,
        )

    def _handle_close_stale_communication(self, session, task: SalesAutomationTaskRow) -> None:
        data = _payload(task)
        communication_id = str(data.get("communication_id") or "").strip()
        if not communication_id:
            raise RuntimeError("communication_id is required")
        communication = session.execute(
            select(SalesCommunicationSessionRow).where(
                SalesCommunicationSessionRow.tenant_id == task.tenant_id,
                SalesCommunicationSessionRow.deal_id == task.deal_id,
                SalesCommunicationSessionRow.communication_id == communication_id,
            )
        ).scalar_one_or_none()
        if communication is None:
            raise RuntimeError(f"Communication not found: {communication_id}")
        if communication.status != "completed":
            now = utc_now_iso()
            communication.status = "completed"
            communication.ended_at = communication.ended_at or now
            communication.result_code = communication.result_code or str(data.get("result_code") or "stale_closed")
            communication.updated_at = now
        follow_up = self._create_task_once(
            session,
            tenant_id=task.tenant_id,
            deal_id=task.deal_id,
            task_type="follow_up_customer",
            payload={"communication_id": communication.communication_id, "source_task_id": task.task_id, "reason": "stale_communication"},
            match_payload={"communication_id": communication.communication_id, "reason": "stale_communication"},
        )
        data["automation_result"] = {"status": "communication_closed", "follow_up_task_id": follow_up.task_id}
        _set_payload(task, data)

    def _handle_escalate_to_human(self, session, task: SalesAutomationTaskRow) -> None:
        deal = self._get_deal(session, task)
        data = _payload(task)
        escalation_type = str(data.get("escalation_type") or "automation_escalation").strip()
        reason = str(data.get("reason") or "automation.escalate_to_human").strip()
        assigned_to = str(data.get("assigned_to_user_id") or deal.assigned_human_user_id or "").strip() or None
        existing = session.execute(
            select(SalesEscalationRow)
            .where(
                SalesEscalationRow.tenant_id == task.tenant_id,
                SalesEscalationRow.deal_id == task.deal_id,
                SalesEscalationRow.escalation_type == escalation_type,
                SalesEscalationRow.status.in_(["open", "assigned", "in_progress"]),
            )
            .order_by(SalesEscalationRow.id.desc())
        ).scalars().first()
        now = utc_now_iso()
        if existing is None:
            existing = SalesEscalationRow(
                escalation_id=new_id("esc"),
                tenant_id=task.tenant_id,
                deal_id=task.deal_id,
                escalation_type=escalation_type,
                reason=reason,
                severity=str(data.get("severity") or "medium"),
                status="assigned" if assigned_to else "open",
                assigned_to_user_id=assigned_to,
                assigned_at=now if assigned_to else None,
                started_at=None,
                created_at=now,
                resolved_at=None,
                canceled_at=None,
                resolution_code=None,
                resolution_summary=None,
                sla_due_at=data.get("sla_due_at"),
                source_channel=deal.current_channel,
                source_communication_session_id=data.get("communication_id"),
                source_automation_task_id=task.task_id,
                updated_at=now,
            )
            session.add(existing)
            self._publish(
                session,
                task=task,
                event_type=sales_event_types.ESCALATION_CREATED,
                aggregate_type="escalation",
                aggregate_id=existing.escalation_id,
                payload={
                    "escalation_id": existing.escalation_id,
                    "escalation_type": existing.escalation_type,
                    "status": existing.status,
                    "severity": existing.severity,
                },
            )
        else:
            existing.updated_at = now
        deal.assigned_human_user_id = assigned_to
        deal.scenario_type = "custom_human_escalation"
        deal.updated_at = now
        if self._stage_code(session, deal) != "transferred_to_support":
            transition_deal_stage(
                session,
                tenant_id=task.tenant_id,
                deal_id=task.deal_id,
                target_stage_code="transferred_to_support",
                actor_type="system",
                reason="automation.escalate_to_human",
                metadata={"automation_task_id": task.task_id, "escalation_id": existing.escalation_id},
                force=True,
            )
        data["automation_result"] = {"status": "escalated", "escalation_id": existing.escalation_id}
        _set_payload(task, data)
        self._publish(
            session,
            task=task,
            event_type=sales_event_types.DEAL_ESCALATION_REQUESTED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            payload={
                "escalation_id": existing.escalation_id,
                "escalation_type": existing.escalation_type,
                "severity": existing.severity,
                "assigned_to_user_id": existing.assigned_to_user_id,
            },
        )

    def _handle_post_sale_transfer(self, session, task: SalesAutomationTaskRow) -> None:
        deal = self._get_deal(session, task)
        current_stage = self._stage_code(session, deal)
        data = _payload(task)
        if current_stage == "transferred_to_execution":
            data["automation_result"] = {"status": "skipped", "reason": "already_transferred"}
            _set_payload(task, data)
            return
        if deal.status != "won":
            data["automation_result"] = {"status": "skipped", "reason": f"deal_status_{deal.status}"}
            _set_payload(task, data)
            return
        transition_deal_stage(
            session,
            tenant_id=task.tenant_id,
            deal_id=task.deal_id,
            target_stage_code="transferred_to_execution",
            actor_type="system",
            reason="automation.post_sale_transfer",
            metadata={"automation_task_id": task.task_id},
            force=True,
        )
        data["automation_result"] = {"status": "transferred_post_sale"}
        _set_payload(task, data)
        self._publish(
            session,
            task=task,
            event_type=sales_event_types.DEAL_TRANSFERRED_POST_SALE,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            payload={"transfer_status": "stub_created"},
        )
