from __future__ import annotations import asyncio import uuid from dataclasses import dataclass, field from typing import Any from sqlalchemy import bindparam, text from sqlalchemy.dialects.postgresql import JSONB from app.config import Settings from app.crm import CrmClient, CrmOutcome, CrmResult from app.domain import ( ContactCandidate, choose_newest, full_jitter_delay, parse_crm_datetime, safe_hash, select_email, validate_phone, ) from app.repository import LeasedTask, LeasedWebhook, Profile, Repository INSERT_CRM_COMMAND = text( """ INSERT INTO bitrix_sync.crm_commands (id,workflow_id,command_type,safe_request,status,attempt_count, next_attempt_at,created_at,updated_at) VALUES (:id,:workflow_id,:type,:safe_request,'in_flight',1,now(),now(),now()) """ ).bindparams(bindparam("safe_request", type_=JSONB())) UPDATE_CRM_COMMAND = text( """ UPDATE bitrix_sync.crm_commands SET status=:status,safe_error_code=:error,http_status=:http_status, safe_response=:safe_response, completed_at=CASE WHEN :terminal THEN now() END, updated_at=now() WHERE id=:id """ ).bindparams(bindparam("safe_response", type_=JSONB())) class BusinessConflict(Exception): def __init__(self, code: str, candidates: list[str] | None = None) -> None: self.code = code self.candidates = candidates or [] super().__init__(code) @dataclass class WorkflowEngine: repository: Repository crm: CrmClient settings: Settings _in_flight: asyncio.Semaphore = field(init=False, repr=False) def __post_init__(self) -> None: self._in_flight = asyncio.Semaphore(self.settings.max_in_flight) async def process(self, task: LeasedTask) -> None: workflow_id = await self.repository.create_workflow(task) profile = await self.repository.load_profile(task.user_id) if profile is None: await self._manual(workflow_id, "profile_missing", task.user_id) await self.repository.complete_task(task, workflow_id) return try: if task.task_type == "contact.map_or_create": await self._map_or_create(workflow_id, profile) elif task.task_type == "contact.update": await self._update(workflow_id, profile) elif task.task_type == "contact.deactivate": await self._deactivate(workflow_id, profile) else: await self._technical_failure(workflow_id, "unknown_task_type") await self.repository.complete_task(task, workflow_id) except BusinessConflict as exc: await self._alert(workflow_id, task.user_id, exc.code, exc.candidates) await self._manual(workflow_id, exc.code, task.user_id) await self.repository.complete_task(task, workflow_id) except RetryableWorkflow as exc: delay = exc.retry_after or full_jitter_delay( task.attempt_count + 1, self.settings.retry_base_seconds, self.settings.retry_max_seconds, ) await self.repository.retry_task(task, exc.code, delay) async def process_webhook(self, item: LeasedWebhook) -> None: if item.receiver_type == "alert": await self.repository.complete_webhook(item) return user_id = await self.repository.mapped_user_for_external(item.external_id) if user_id is None: await self.repository.complete_webhook(item) return workflow_id = uuid.uuid4() async with self.repository.transaction() as connection: await connection.execute( text( """ INSERT INTO bitrix_sync.workflow_instances (id,workflow_type,user_id,external_id,state,current_step,deadline_at, created_at,updated_at) VALUES (:id,'contact.webhook',:user_id,:external_id,'running', 'read_contact',now()+interval '24 hours',now(),now()) """ ), {"id": workflow_id, "user_id": user_id, "external_id": item.external_id}, ) try: result = await self._command( workflow_id, "contact_get", "crm.contact.get", {"id": item.external_id, "select": self._select_fields()}, mutating=False, ) contact = result.result if str(contact.get(self.settings.contact_user_id_field)) != str(user_id): await self._alert( workflow_id, user_id, "mapping_identity_mismatch", [item.external_id], ) await self._manual(workflow_id, "mapping_identity_mismatch", user_id) await self.repository.complete_webhook(item) return citizenship = await self._resolve_citizenship( workflow_id, contact.get(self.settings.contact_citizenship_field) ) source_value = contact.get("DATE_MODIFY") or contact.get("updatedTime") await self.repository.apply_crm_profile( user_id, item.external_id, full_name=contact.get("NAME") or None, citizenship=citizenship, email=select_email(contact.get("EMAIL") or []), source_updated_at=parse_crm_datetime(source_value) if source_value else None, source="reconciliation" if item.event_type == "contact.reconciliation" else "webhook", ) await self.repository.complete_webhook(item) async with self.repository.transaction() as connection: await connection.execute( text( """ UPDATE bitrix_sync.workflow_instances SET state='succeeded',current_step='done', completed_at=now(),updated_at=now() WHERE id=:id """ ), {"id": workflow_id}, ) except RetryableWorkflow: raise async def _map_or_create(self, workflow_id: uuid.UUID, profile: Profile) -> None: if profile.identity_status != "A" or profile.profile_status != "A": await self._deactivate(workflow_id, profile) return validate_phone(profile.phone) if mapped := await self.repository.active_mapping(profile.user_id): await self._ensure_registered(workflow_id, mapped, profile.user_id) return found = await self._command( workflow_id, "duplicate_find", "crm.duplicate.findbycomm", {"type": "PHONE", "values": [profile.phone], "entity_type": "CONTACT"}, mutating=False, ) ids = _contact_ids(found.result) contacts: list[dict[str, Any]] = [] for contact_id in ids: result = await self._command( workflow_id, "contact_get", "crm.contact.get", {"id": contact_id, "select": self._select_fields()}, mutating=False, ) if isinstance(result.result, dict): contacts.append(result.result) same_user = next( ( item for item in contacts if str(item.get(self.settings.contact_user_id_field)) == str(profile.user_id) ), None, ) alert_code: str | None = None if same_user: selected_id = str(same_user["ID"]) if len(contacts) > 1: await self._alert( workflow_id, profile.user_id, "duplicate_contacts", [str(item["ID"]) for item in contacts], selected_external_id=selected_id, ) elif not contacts: selected_id = await self._create_contact(workflow_id, profile) else: candidates = [ ContactCandidate( str(item["ID"]), parse_crm_datetime(item.get("CREATED_TIME", "1970-01-01T00:00:00Z")), item.get(self.settings.contact_user_id_field), ) for item in contacts ] selected = choose_newest(candidates) assert selected is not None all_ids = [item.b24_id for item in candidates] if selected.crm_user_id and selected.crm_user_id != str(profile.user_id): alert_code = "contact_owned_by_other_user" selected_id = await self._create_contact(workflow_id, profile) await self._alert( workflow_id, profile.user_id, alert_code, [*all_ids, selected_id], selected_external_id=selected_id, ) else: selected_id = selected.b24_id if len(candidates) > 1: alert_code = "duplicate_contacts" await self._alert( workflow_id, profile.user_id, alert_code, all_ids, selected_external_id=selected_id, ) await self._write_identity(workflow_id, selected_id, profile.user_id, active=True) await self._activate_mapping(workflow_id, profile.user_id, selected_id) async def _create_contact(self, workflow_id: uuid.UUID, profile: Profile) -> str: result = await self._command( workflow_id, "contact_add", "crm.contact.add", { "fields": { "PHONE": [{"VALUE": profile.phone, "VALUE_TYPE": "WORK"}], self.settings.contact_user_id_field: str(profile.user_id), self.settings.contact_registered_field: "1", } }, mutating=True, ) return str(result.result) async def _update(self, workflow_id: uuid.UUID, profile: Profile) -> None: validate_phone(profile.phone) mapping = await self.repository.active_mapping(profile.user_id) if mapping is None: await self._coalesce_map_or_create(profile.user_id) return await self._command( workflow_id, "contact_update", "crm.contact.update", { "id": mapping, "fields": {"PHONE": [{"VALUE": profile.phone, "VALUE_TYPE": "WORK"}]}, }, mutating=True, ) async def _deactivate(self, workflow_id: uuid.UUID, profile: Profile) -> None: mapping = await self.repository.active_mapping(profile.user_id) if mapping is None: return await self._write_identity(workflow_id, mapping, profile.user_id, active=False) async with self.repository.transaction() as connection: await connection.execute( text( """ UPDATE bitrix_sync.entity_external_mapping SET status='closed', closed_at=now(), close_reason='deactivated', workflow_id=:workflow_id, updated_at=now() WHERE entity_id=:user_id AND external_id=:external_id AND status='active' """ ), { "workflow_id": workflow_id, "user_id": profile.user_id, "external_id": mapping, }, ) async def process_rebind(self, request_id: uuid.UUID) -> None: async with self.repository.transaction() as connection: request = ( await connection.execute( text( """ SELECT id,user_id,old_external_id,target_external_id,workflow_id FROM bitrix_sync.rebind_requests WHERE id=:id AND status IN ('pending','retry_wait') FOR UPDATE """ ), {"id": request_id}, ) ).mappings().first() if not request: return await connection.execute( text( """ UPDATE bitrix_sync.rebind_requests SET status='processing',updated_at=now() WHERE id=:id """ ), {"id": request_id}, ) target = await self._command( request["workflow_id"], "rebind_target_get", "crm.contact.get", {"id": request["target_external_id"], "select": self._select_fields()}, mutating=False, ) target_user = target.result.get(self.settings.contact_user_id_field) if target_user and target_user != str(request["user_id"]): raise BusinessConflict("rebind_target_owned", [request["target_external_id"]]) await self._write_identity( request["workflow_id"], request["target_external_id"], request["user_id"], active=True ) if request["old_external_id"]: old = await self._command( request["workflow_id"], "rebind_old_get", "crm.contact.get", {"id": request["old_external_id"], "select": self._select_fields()}, mutating=False, ) if old.result.get(self.settings.contact_user_id_field) == str(request["user_id"]): await self._write_identity( request["workflow_id"], request["old_external_id"], request["user_id"], active=False, ) async with self.repository.transaction() as connection: await connection.execute( text( """ UPDATE bitrix_sync.entity_external_mapping SET status='closed',closed_at=now(),close_reason='rebind', workflow_id=:workflow_id,updated_at=now() WHERE entity_id=:user_id AND status='active' """ ), { "workflow_id": request["workflow_id"], "user_id": request["user_id"], }, ) await connection.execute( text( """ INSERT INTO bitrix_sync.entity_external_mapping (id,entity_type,entity_id,external_system,external_entity_type, external_id,status,opened_at,workflow_id,created_at,updated_at) VALUES (gen_random_uuid(),'contact',:user_id,'bitrix24','contact', :target,'active',now(),:workflow_id,now(),now()) """ ), { "workflow_id": request["workflow_id"], "user_id": request["user_id"], "target": request["target_external_id"], }, ) await connection.execute( text( """ UPDATE bitrix_sync.rebind_requests SET status='succeeded',completed_at=now(),updated_at=now() WHERE id=:request_id """ ), { "request_id": request_id, }, ) async def _ensure_registered( self, workflow_id: uuid.UUID, external_id: str, user_id: uuid.UUID ) -> None: contact = await self._command( workflow_id, "contact_get", "crm.contact.get", {"id": external_id, "select": self._select_fields()}, mutating=False, ) if str(contact.result.get(self.settings.contact_registered_field)) not in {"1", "Y"}: await self._write_identity(workflow_id, external_id, user_id, active=True) async def _resolve_citizenship( self, workflow_id: uuid.UUID, enum_id: str | int | None ) -> str | None: if enum_id in (None, ""): return None async with self.repository.transaction() as connection: value = ( await connection.execute( text( """ SELECT display_value FROM bitrix_sync.citizenship_dictionary WHERE enum_id=:enum_id AND expires_at>now() """ ), {"enum_id": str(enum_id)}, ) ).scalar_one_or_none() if value is not None: return value result = await self._command( workflow_id, "citizenship_fields_get", "crm.contact.userfield.list", {"filter": {"FIELD_NAME": self.settings.contact_citizenship_field}}, mutating=False, ) fields = result.result if isinstance(result.result, list) else [] entries = fields[0].get("LIST", []) if fields else [] async with self.repository.transaction() as connection: for entry in entries: if entry.get("ID") is None or not isinstance(entry.get("VALUE"), str): continue await connection.execute( text( """ INSERT INTO bitrix_sync.citizenship_dictionary (enum_id,display_value,loaded_at,expires_at) VALUES (:id,:value,now(),now()+interval '1 hour') ON CONFLICT (enum_id) DO UPDATE SET display_value=excluded.display_value,loaded_at=now(), expires_at=excluded.expires_at """ ), {"id": str(entry["ID"]), "value": entry["VALUE"]}, ) match = next( ( entry["VALUE"] for entry in entries if str(entry.get("ID")) == str(enum_id) ), None, ) if match is None: raise BusinessConflict("unknown_citizenship_enum", [str(enum_id)]) return match async def _write_identity( self, workflow_id: uuid.UUID, external_id: str, user_id: uuid.UUID, *, active: bool ) -> None: fields: dict[str, Any] = {self.settings.contact_registered_field: "1" if active else "0"} fields[self.settings.contact_user_id_field] = str(user_id) if active else "" await self._command( workflow_id, "contact_update", "crm.contact.update", {"id": external_id, "fields": fields}, mutating=True, ) async def _command( self, workflow_id: uuid.UUID, command_type: str, method: str, params: dict[str, Any], *, mutating: bool, ) -> CrmResult: command_id = uuid.uuid4() async with self.repository.transaction() as connection: await connection.execute( INSERT_CRM_COMMAND, { "id": command_id, "workflow_id": workflow_id, "type": command_type, "safe_request": {"keys": sorted(params), "method_class": method.split(".")[-1]}, }, ) while True: limiter_delay = await self.repository.reserve_limiter_token( self.settings.limiter_refill_per_sec, self.settings.limiter_burst ) if limiter_delay <= 0: break await asyncio.sleep(limiter_delay) async with self._in_flight: result = await self.crm.call(method, params, mutating=mutating) status = result.outcome.value async with self.repository.transaction() as connection: await connection.execute( UPDATE_CRM_COMMAND, { "id": command_id, "status": status, "error": result.error_code, "http_status": result.http_status, "safe_response": {"has_result": result.result is not None}, "terminal": result.outcome in {CrmOutcome.SUCCEEDED, CrmOutcome.PERMANENT}, }, ) if result.outcome == CrmOutcome.SUCCEEDED: return result if result.outcome == CrmOutcome.PERMANENT: await self._technical_failure(workflow_id, result.error_code or "crm_permanent") raise BusinessConflict("technical_configuration_failure") raise RetryableWorkflow(result.error_code or result.outcome.value, result.retry_after) async def _activate_mapping( self, workflow_id: uuid.UUID, user_id: uuid.UUID, external_id: str ) -> None: async with self.repository.transaction() as connection: await connection.execute( text( """ INSERT INTO bitrix_sync.entity_external_mapping (id,entity_type,entity_id,external_system,external_entity_type, external_id,status,opened_at,workflow_id,created_at,updated_at) VALUES (gen_random_uuid(),'contact',:user_id,'bitrix24','contact', :external_id,'active',now(),:workflow_id,now(),now()) ON CONFLICT DO NOTHING """ ), { "workflow_id": workflow_id, "user_id": user_id, "external_id": external_id, }, ) async def _coalesce_map_or_create(self, user_id: uuid.UUID) -> None: async with self.repository.transaction() as connection: await connection.execute( text( """ INSERT INTO han_app.sync_queue (id,task_type,entity_type,entity_id,dedup_key,payload_json,status, attempt_count,next_attempt_at,created_at,updated_at) VALUES (gen_random_uuid(),'contact.map_or_create','contact',:user_id, 'contact.map_or_create:'||:user_id::text, jsonb_build_object('schema_version',1,'user_id',:user_id), 'pending',0,now(),now(),now()) ON CONFLICT DO NOTHING """ ), {"user_id": user_id}, ) async def _alert( self, workflow_id: uuid.UUID, user_id: uuid.UUID, alert_type: str, candidates: list[str], *, selected_external_id: str | None = None, ) -> None: fingerprint = safe_hash(f"{alert_type}:{user_id}") assert fingerprint is not None async with self.repository.transaction() as connection: alert = ( await connection.execute( text( """ INSERT INTO bitrix_sync.business_alerts (id,fingerprint,alert_type,severity,app_user_id, selected_external_id,candidate_external_ids,workflow_id,status, occurrence_count,first_occurred_at,last_occurred_at,created_at,updated_at) VALUES (gen_random_uuid(),:fingerprint, :type,'warning',:user_id,:selected_external_id,:candidates, :workflow_id,'open',1,now(),now(),now(),now()) ON CONFLICT (alert_type,fingerprint) WHERE status='open' DO UPDATE SET selected_external_id=coalesce( excluded.selected_external_id, business_alerts.selected_external_id ), candidate_external_ids=excluded.candidate_external_ids, workflow_id=excluded.workflow_id, occurrence_count=business_alerts.occurrence_count+1, last_occurred_at=now(),updated_at=now() RETURNING id,alert_number,fingerprint,alert_type,severity,app_user_id, selected_external_id,candidate_external_ids,remote_item_id, occurrence_count,first_occurred_at,last_occurred_at """ ), { "fingerprint": fingerprint, "type": alert_type, "user_id": user_id, "selected_external_id": selected_external_id, "candidates": candidates, "workflow_id": workflow_id, }, ) ).mappings().one() alert_config = ( await connection.execute( text( """ SELECT value_json FROM bitrix_sync.settings WHERE key='business_alerts' AND active=true AND validation_status='valid' """ ) ) ).scalar_one_or_none() if not isinstance(alert_config, dict): raise RetryableWorkflow("business_alerts_not_configured") entity_type_id = alert_config.get("entity_type_id") stage_new = alert_config.get("stage_new") if not entity_type_id or not stage_new: raise RetryableWorkflow("business_alerts_not_configured") title = ( f"Конфликт синхронизации #{alert['alert_number']}: {alert_type}; " f"user={user_id}; contacts={','.join(candidates)}" ) fields: dict[str, Any] = { "title": title[:255], "stageId": stage_new, "contactIds": [int(candidate) for candidate in candidates if candidate.isdigit()], "sourceId": self.settings.contact_source, } if category_id := alert_config.get("category_id"): fields["categoryId"] = category_id if responsible_id := alert_config.get("responsible_id"): fields["assignedById"] = responsible_id alert_values = { "alert_number": str(alert["alert_number"]), "fingerprint": alert["fingerprint"], "alert_type": alert["alert_type"], "severity": alert["severity"], "app_user_id": str(alert["app_user_id"]), "selected_external_id": alert["selected_external_id"] or "", "occurrence_count": alert["occurrence_count"], "first_occurred_at": alert["first_occurred_at"].isoformat(), "last_occurred_at": alert["last_occurred_at"].isoformat(), "workflow_id": str(workflow_id), } field_ids = alert_config.get("field_ids") if isinstance(field_ids, dict): for value_name, field_id in field_ids.items(): if value_name in alert_values and isinstance(field_id, str) and field_id: fields[field_id] = alert_values[value_name] remote_item_id = alert["remote_item_id"] if remote_item_id: await self._command( workflow_id, "alert_update", "crm.item.update", { "entityTypeId": int(entity_type_id), "id": remote_item_id, "fields": fields, }, mutating=True, ) return result = await self._command( workflow_id, "alert_add", "crm.item.add", {"entityTypeId": int(entity_type_id), "fields": fields}, mutating=True, ) payload = result.result if isinstance(result.result, dict) else {} item = payload.get("item", payload) remote_item_id = item.get("id") if isinstance(item, dict) else None if remote_item_id is None: raise RetryableWorkflow("alert_item_id_missing") async with self.repository.transaction() as connection: await connection.execute( text( """ UPDATE bitrix_sync.business_alerts SET remote_item_id=:remote_item_id,remote_stage_id=:stage_id,updated_at=now() WHERE id=:id """ ), { "id": alert["id"], "remote_item_id": str(remote_item_id), "stage_id": str(stage_new), }, ) async def _manual(self, workflow_id: uuid.UUID, code: str, user_id: uuid.UUID) -> None: async with self.repository.transaction() as connection: await connection.execute( text( """ UPDATE bitrix_sync.workflow_instances SET state='waiting_manual',outcome=:code,updated_at=now() WHERE id=:id """ ), {"id": workflow_id, "code": code}, ) async def _technical_failure(self, workflow_id: uuid.UUID, code: str) -> None: async with self.repository.transaction() as connection: await connection.execute( text( """ INSERT INTO bitrix_sync.technical_dead_letters (id,workflow_id,operation,safe_error_code,failed_at,created_at) VALUES (gen_random_uuid(),:workflow_id,'crm_command',:code,now(),now()) """ ), {"workflow_id": workflow_id, "code": code[:64]}, ) def _select_fields(self) -> list[str]: return [ "ID", "NAME", "PHONE", "EMAIL", "CREATED_TIME", "DATE_MODIFY", self.settings.contact_user_id_field, self.settings.contact_registered_field, self.settings.contact_citizenship_field, ] class RetryableWorkflow(Exception): def __init__(self, code: str, retry_after: float | None = None) -> None: self.code = code self.retry_after = retry_after super().__init__(code) def _contact_ids(result: Any) -> list[str]: if isinstance(result, dict): values = result.get("CONTACT", []) else: values = result or [] return sorted({str(value) for value in values if str(value).isdigit()}, key=int)