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"}, )