from __future__ import annotations import os import ssl import uuid from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine from app.domain import safe_hash from app.telemetry import record_status_snapshot def postgres_ssl_context() -> ssl.SSLContext: ca_file = os.environ.get("PG_CA_FILE") if not ca_file: raise RuntimeError("PG_CA_FILE is required") context = ssl.create_default_context(cafile=ca_file) context.check_hostname = True context.verify_mode = ssl.CERT_REQUIRED return context @dataclass(frozen=True) class LeasedTask: id: uuid.UUID task_type: str user_id: uuid.UUID lease_token: uuid.UUID attempt_count: int @dataclass(frozen=True) class LeasedWebhook: id: uuid.UUID receiver_type: str event_type: str external_id: str lease_token: uuid.UUID attempt_count: int @dataclass(frozen=True) class Profile: user_id: uuid.UUID phone: str identity_status: str profile_status: str class Repository: def __init__(self, database_url: str, pool_size: int = 5) -> None: self.engine: AsyncEngine = create_async_engine( database_url, pool_size=pool_size, pool_pre_ping=True, connect_args={"ssl": postgres_ssl_context()}, ) async def close(self) -> None: await self.engine.dispose() @asynccontextmanager async def transaction(self) -> AsyncIterator[AsyncConnection]: async with self.engine.begin() as connection: yield connection async def ping(self) -> bool: try: async with self.engine.connect() as connection: await connection.execute(text("SELECT 1")) return True except Exception: return False async def claim_tasks( self, worker_id: str, limit: int, lease_seconds: int ) -> list[LeasedTask]: sql = text( """ WITH candidates AS ( SELECT id FROM han_app.sync_queue WHERE ( status IN ('pending','retry_wait') AND next_attempt_at <= now() ) OR ( status='leased' AND locked_until < now() ) ORDER BY next_attempt_at, created_at FOR UPDATE SKIP LOCKED LIMIT :limit ) UPDATE han_app.sync_queue q SET status='leased', locked_by=:worker_id, locked_until=now() + make_interval(secs => :lease_seconds), lease_token=gen_random_uuid(), updated_at=now() FROM candidates c WHERE q.id=c.id RETURNING q.id, q.task_type, q.entity_id, q.lease_token, q.attempt_count """ ) async with self.engine.begin() as connection: rows = ( await connection.execute( sql, {"worker_id": worker_id, "limit": limit, "lease_seconds": lease_seconds}, ) ).mappings() return [ LeasedTask( row["id"], row["task_type"], row["entity_id"], row["lease_token"], row["attempt_count"], ) for row in rows ] async def load_profile(self, user_id: uuid.UUID) -> Profile | None: sql = text( """ SELECT i.id user_id, i.phone_number phone, i.record_status identity_status, p.record_status profile_status FROM han_app.user_identities i JOIN han_app.client_profiles p ON p.user_id=i.id WHERE i.id=:user_id """ ) async with self.engine.connect() as connection: row = (await connection.execute(sql, {"user_id": user_id})).mappings().first() return Profile(**row) if row else None async def active_mapping(self, user_id: uuid.UUID) -> str | None: sql = text( """ SELECT external_id FROM bitrix_sync.entity_external_mapping WHERE external_system='bitrix24' AND entity_type='contact' AND entity_id=:user_id AND status='active' """ ) async with self.engine.connect() as connection: return (await connection.execute(sql, {"user_id": user_id})).scalar_one_or_none() async def create_workflow(self, task: LeasedTask) -> uuid.UUID: workflow_id = uuid.uuid4() sql = text( """ INSERT INTO bitrix_sync.workflow_instances (id, workflow_type, user_id, state, current_step, source_task_id, deadline_at, created_at, updated_at) VALUES (:id, :workflow_type, :user_id, 'created', 'load_profile', :task_id, now() + interval '24 hours', now(), now()) ON CONFLICT (source_task_id) DO UPDATE SET updated_at=now() RETURNING id """ ) async with self.engine.begin() as connection: workflow_id = ( await connection.execute( sql, { "id": workflow_id, "workflow_type": task.task_type, "user_id": task.user_id, "task_id": task.id, }, ) ).scalar_one() await connection.execute( text( """ UPDATE bitrix_sync.workflow_instances SET state='running',current_step='load_profile',updated_at=now() WHERE id=:id AND state IN ('created','running') """ ), {"id": workflow_id}, ) return workflow_id async def complete_task(self, task: LeasedTask, workflow_id: uuid.UUID) -> bool: async with self.engine.begin() as connection: result = await connection.execute( text( """ UPDATE han_app.sync_queue SET status='processed', completed_at=now(), locked_by=NULL, locked_until=NULL, lease_token=NULL, updated_at=now() WHERE id=:id AND status='leased' AND lease_token=:lease_token """ ), {"id": task.id, "lease_token": task.lease_token}, ) if result.rowcount != 1: return False await connection.execute( text( """ UPDATE bitrix_sync.workflow_instances SET state='succeeded', current_step='done', outcome='processed', completed_at=now(), updated_at=now() WHERE id=:workflow_id AND state NOT IN ('succeeded','failed','cancelled') """ ), {"workflow_id": workflow_id}, ) return True async def retry_task( self, task: LeasedTask, safe_code: str, delay_seconds: float ) -> bool: sql = text( """ UPDATE han_app.sync_queue SET status='retry_wait', attempt_count=attempt_count+1, next_attempt_at=now() + make_interval(secs => :delay), last_error_code=:code, last_error_at=now(), locked_by=NULL, locked_until=NULL, lease_token=NULL, updated_at=now() WHERE id=:id AND status='leased' AND lease_token=:lease_token """ ) async with self.engine.begin() as connection: result = await connection.execute( sql, { "id": task.id, "lease_token": task.lease_token, "code": safe_code[:64], "delay": delay_seconds, }, ) return result.rowcount == 1 async def insert_webhook( self, receiver_type: str, event_type: str, entity_id: str, source_ip: str, ) -> uuid.UUID: inbox_id = uuid.uuid4() async with self.engine.begin() as connection: existing = await connection.execute( text( """ SELECT id FROM bitrix_sync.webhook_inbox WHERE receiver_type=:receiver AND external_entity_id=:entity_id AND status IN ('received','processing','retry_wait') FOR UPDATE """ ), {"receiver": receiver_type, "entity_id": entity_id}, ) if row := existing.first(): await connection.execute( text( """ UPDATE bitrix_sync.webhook_inbox SET coalesced_count=coalesced_count+1, last_received_at=now() WHERE id=:id """ ), {"id": row[0]}, ) return row[0] await connection.execute( text( """ INSERT INTO bitrix_sync.webhook_inbox (id, receiver_type, event_type, external_entity_id, source_ip, status, coalesced_count, received_at, last_received_at) VALUES (:id,:receiver,:event,:entity_id,CAST(:source_ip AS inet), 'received',1,now(),now()) """ ), { "id": inbox_id, "receiver": receiver_type, "event": event_type, "entity_id": entity_id, "source_ip": source_ip, }, ) return inbox_id async def claim_webhooks( self, worker_id: str, limit: int, lease_seconds: int ) -> list[LeasedWebhook]: sql = text( """ WITH candidates AS ( SELECT id FROM bitrix_sync.webhook_inbox WHERE ( status IN ('received','retry_wait') AND next_attempt_at<=now() ) OR ( status='processing' AND locked_until:lease_seconds), lease_token=gen_random_uuid() FROM candidates c WHERE w.id=c.id RETURNING w.id,w.receiver_type,w.event_type,w.external_entity_id, w.lease_token,w.attempt_count """ ) async with self.engine.begin() as connection: rows = ( await connection.execute( sql, {"worker_id": worker_id, "limit": limit, "lease_seconds": lease_seconds}, ) ).mappings() return [ LeasedWebhook( row["id"], row["receiver_type"], row["event_type"], row["external_entity_id"], row["lease_token"], row["attempt_count"], ) for row in rows ] async def mapped_user_for_external(self, external_id: str) -> uuid.UUID | None: async with self.engine.connect() as connection: return ( await connection.execute( text( """ SELECT entity_id FROM bitrix_sync.entity_external_mapping WHERE external_system='bitrix24' AND external_entity_type='contact' AND external_id=:external_id AND status='active' """ ), {"external_id": external_id}, ) ).scalar_one_or_none() async def apply_crm_profile( self, user_id: uuid.UUID, external_id: str, *, full_name: str | None, citizenship: str | None, email: str | None, source_updated_at: datetime | None, source: str, ) -> None: async with self.engine.begin() as connection: await connection.execute(text("SET LOCAL han.sync_suppress='true'")) await connection.execute( text( """ UPDATE han_app.client_profiles SET full_name=:full_name,citizenship=:citizenship,email=:email, source_updated_at=:source_updated_at,updated_at=now() WHERE user_id=:user_id AND record_status='A' """ ), { "user_id": user_id, "full_name": full_name, "citizenship": citizenship, "email": email, "source_updated_at": source_updated_at, }, ) await connection.execute( text( """ INSERT INTO bitrix_sync.contact_snapshots (id,mapping_id,user_id,external_id,full_name_hash,email_hash, citizenship_hash,source_updated_at,last_applied_source, last_webhook_received_at,created_at,updated_at) SELECT gen_random_uuid(),m.id,:user_id, CAST(:external_id AS varchar(128)), CAST(:full_name_hash AS varchar(64)), CAST(:email_hash AS varchar(64)), CAST(:citizenship_hash AS varchar(64)), CAST(:source_updated_at AS timestamptz), CAST(:source AS varchar(24)), CASE WHEN CAST(:source AS varchar(24))='webhook' THEN now() END, now(),now() FROM bitrix_sync.entity_external_mapping m WHERE m.entity_id=:user_id AND m.external_id=CAST(:external_id AS varchar(128)) AND m.status='active' ON CONFLICT (mapping_id) DO UPDATE SET full_name_hash=excluded.full_name_hash,email_hash=excluded.email_hash, citizenship_hash=excluded.citizenship_hash, source_updated_at=excluded.source_updated_at, last_applied_source=excluded.last_applied_source, last_webhook_received_at=coalesce( excluded.last_webhook_received_at, bitrix_sync.contact_snapshots.last_webhook_received_at), updated_at=now() """ ), { "user_id": user_id, "external_id": external_id, "full_name_hash": safe_hash(full_name or ""), "citizenship_hash": safe_hash(citizenship or ""), "email_hash": safe_hash(email or ""), "source_updated_at": source_updated_at, "source": source, }, ) async def complete_webhook(self, item: LeasedWebhook) -> bool: async with self.engine.begin() as connection: result = await connection.execute( text( """ UPDATE bitrix_sync.webhook_inbox SET status='processed',processed_at=now(),locked_by=NULL, locked_until=NULL,lease_token=NULL WHERE id=:id AND status='processing' AND lease_token=:lease_token """ ), {"id": item.id, "lease_token": item.lease_token}, ) return result.rowcount == 1 async def retry_webhook( self, item: LeasedWebhook, safe_code: str, delay_seconds: float ) -> bool: async with self.engine.begin() as connection: result = await connection.execute( text( """ UPDATE bitrix_sync.webhook_inbox SET status='retry_wait',attempt_count=attempt_count+1, next_attempt_at=now()+make_interval(secs=>:delay), safe_error_code=:code,locked_by=NULL,locked_until=NULL,lease_token=NULL WHERE id=:id AND status='processing' AND lease_token=:lease_token """ ), { "id": item.id, "lease_token": item.lease_token, "code": safe_code[:64], "delay": delay_seconds, }, ) return result.rowcount == 1 async def pending_rebind_ids(self, limit: int) -> list[uuid.UUID]: async with self.engine.connect() as connection: rows = await connection.execute( text( """ SELECT id FROM bitrix_sync.rebind_requests WHERE status IN ('pending','retry_wait') ORDER BY requested_at LIMIT :limit """ ), {"limit": limit}, ) return list(rows.scalars()) async def reserve_limiter_token(self, refill_per_second: float, burst: int) -> float: async with self.engine.begin() as connection: row = ( await connection.execute( text( """ SELECT tokens,capacity,refill_per_second, extract(epoch FROM now()-updated_at) elapsed, greatest(0,extract(epoch FROM blocked_until-now())) blocked FROM bitrix_sync.limiter_coordination WHERE limiter_key='bitrix24:portal' FOR UPDATE """ ) ) ).mappings().first() if row is None: await connection.execute( text( """ INSERT INTO bitrix_sync.limiter_coordination (limiter_key,tokens,capacity,refill_per_second,updated_at,fencing_token) VALUES ('bitrix24:portal',:tokens,:capacity,:refill,now(),1) """ ), {"tokens": max(0, burst - 1), "capacity": burst, "refill": refill_per_second}, ) return 0 blocked = float(row["blocked"] or 0) tokens = min( float(burst), float(row["tokens"]) + float(row["elapsed"] or 0) * refill_per_second, ) if blocked > 0: delay = blocked elif tokens >= 1: tokens -= 1 delay = 0 else: delay = (1 - tokens) / refill_per_second await connection.execute( text( """ UPDATE bitrix_sync.limiter_coordination SET tokens=:tokens,capacity=:capacity,refill_per_second=:refill, updated_at=now(),fencing_token=fencing_token+1 WHERE limiter_key='bitrix24:portal' """ ), { "tokens": tokens, "capacity": burst, "refill": refill_per_second, }, ) return delay async def status(self) -> dict[str, Any]: queries = { "queue": "SELECT status, count(*) count FROM han_app.sync_queue GROUP BY status", "workflows": ( "SELECT state AS status, count(*) count " "FROM bitrix_sync.workflow_instances GROUP BY state" ), "commands": ( "SELECT status, count(*) count " "FROM bitrix_sync.crm_commands GROUP BY status" ), "queue_oldest_age_seconds": ( "SELECT coalesce(extract(epoch from now()-min(created_at)),0) " "FROM han_app.sync_queue WHERE status IN ('pending','retry_wait','leased')" ), "webhook_lag_seconds": ( "SELECT coalesce(extract(epoch from now()-min(received_at)),0) " "FROM bitrix_sync.webhook_inbox WHERE status IN ('received','retry_wait')" ), "settings_version": ( "SELECT version FROM bitrix_sync.settings_versions " "WHERE active=true AND validation_status='valid' ORDER BY activated_at DESC LIMIT 1" ), } output: dict[str, Any] = {} async with self.engine.connect() as connection: for key, sql in queries.items(): result = await connection.execute(text(sql)) if key in {"queue", "workflows", "commands"}: output[key] = {row.status: row.count for row in result} else: output[key] = result.scalar_one_or_none() output["generated_at"] = datetime.now(UTC).isoformat() record_status_snapshot(output) return output