Admin/han-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Поправили заполлнение БД + поправили гонку сообщений при отправке в Битрикс

Browse Source
This commit is contained in:
mi
2026-07-17 11:46:12 +03:00
parent 7457bc5c0b
commit d1a0f3791d
13 changed files with 562 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codebase/backend/api-backend/app/services.py
+250 -48
View File
@@ -9,7 +9,7 @@ from pathlib import PurePath
from typing import Any
import httpx
from sqlalchemy import func, select
from sqlalchemy import case, func, select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import AsyncSession
@@ -106,6 +106,24 @@ class SettingsSnapshot:
return int(amount), windows[period]
@dataclass(frozen=True, slots=True)
class AuditContext:
request_id: str
trace_id: str
ux_session_id: uuid.UUID | None
user_agent_hash: str | None
client_ip: str | None
def with_ux_session(self, ux_session_id: uuid.UUID) -> "AuditContext":
return AuditContext(
request_id=self.request_id,
trace_id=self.trace_id,
ux_session_id=ux_session_id,
user_agent_hash=self.user_agent_hash,
client_ip=self.client_ip,
)
async def load_settings(session: AsyncSession) -> SettingsSnapshot:
rows = (
await session.execute(select(AppSetting).where(AppSetting.record_status == "A"))
@@ -138,26 +156,45 @@ async def resolve_user(session: AsyncSession, principal: Principal) -> UserIdent
def audit(
event_type: str,
request_id: str,
context: AuditContext,
user_id: uuid.UUID | None,
resource_type: str | None = None,
resource_id: uuid.UUID | None = None,
ux_session_id: uuid.UUID | None = None,
metadata: dict[str, Any] | None = None,
outcome: str = "success",
) -> AuditEvent:
return AuditEvent(
event_type=event_type,
actor_type="user" if user_id else "service",
user_id=user_id,
ux_session_id=ux_session_id,
request_id=request_id,
ux_session_id=context.ux_session_id,
request_id=context.request_id,
trace_id=context.trace_id,
resource_type=resource_type,
resource_id=resource_id,
outcome="success",
user_agent_hash=context.user_agent_hash,
outcome=outcome,
metadata_json=metadata or {},
)
def device_snapshot(device: Any) -> dict[str, str | None]:
return {
"platform": device.platform,
"app_version": device.app_version,
"device_id_hash": (
hashlib.sha256(device.device_id.encode()).hexdigest() if device.device_id else None
),
}
def consent_versions(consents: Any) -> dict[str, str]:
return {
name: getattr(consents, name).version
for name in ("personal_data", "user_agreement", "marketing")
}
def validate_consents(consents: Any, snapshot: SettingsSnapshot) -> None:
for name in ("personal_data", "user_agreement", "marketing"):
choice = getattr(consents, name)
@@ -174,12 +211,13 @@ async def bootstrap(
principal: Principal,
body: BootstrapRequest,
snapshot: SettingsSnapshot,
request_id: str,
context: AuditContext,
) -> dict[str, Any]:
if not principal.phone_number:
raise DomainError("phone_claim_missing", 400, "Verified phone claim is missing")
validate_consents(body.consents, snapshot)
now = datetime.now(UTC)
device = device_snapshot(body.device)
statement = (
insert(UserIdentity)
.values(
@@ -211,6 +249,9 @@ async def bootstrap(
document_version=choice.version,
accepted=choice.accepted,
accepted_at=now,
client_ip=context.client_ip,
user_agent_hash=context.user_agent_hash,
device_json=device,
)
.on_conflict_do_nothing(
index_elements=[
@@ -220,7 +261,18 @@ async def bootstrap(
]
)
)
session.add(audit("auth.bootstrap", request_id, user_id))
session.add(
audit(
"auth.bootstrap",
context,
user_id,
metadata={
"consent_versions": consent_versions(body.consents),
"platform": body.device.platform,
"app_version": body.device.app_version,
},
)
)
await session.commit()
return {"user_id": user_id, "profile_ready": True}
@@ -230,42 +282,83 @@ async def record_consents(
user: UserIdentity,
body: ConsentsRequest,
snapshot: SettingsSnapshot,
request_id: str,
ux_session_id: uuid.UUID | None,
context: AuditContext,
) -> dict[str, Any]:
validate_consents(body.consents, snapshot)
if context.ux_session_id is None:
raise DomainError("validation_error", 400, "X-Ux-Session-Id is required")
ux_session = (
await session.execute(
select(UxSession).where(
UxSession.id == context.ux_session_id,
UxSession.user_id == user.id,
)
)
).scalar_one_or_none()
if ux_session is None:
raise DomainError("validation_error", 400, "X-Ux-Session-Id is invalid")
device = {
"platform": ux_session.platform,
"app_version": ux_session.app_version,
"device_id_hash": ux_session.device_id_hash,
}
now = datetime.now(UTC)
versions: dict[str, str] = {}
for consent_type in ("personal_data", "user_agreement", "marketing"):
choice = getattr(body.consents, consent_type)
versions[consent_type] = choice.version
await session.execute(
insert(UserConsent)
.values(
id=uuid.uuid4(),
user_id=user.id,
ux_session_id=ux_session_id,
consent_type=consent_type,
document_version=choice.version,
accepted=choice.accepted,
accepted_at=now,
)
.on_conflict_do_nothing()
statement = insert(UserConsent).values(
id=uuid.uuid4(),
user_id=user.id,
ux_session_id=context.ux_session_id,
consent_type=consent_type,
document_version=choice.version,
accepted=choice.accepted,
accepted_at=now,
client_ip=context.client_ip,
user_agent_hash=context.user_agent_hash,
device_json=device,
)
session.add(audit("consent.recorded", request_id, user.id, ux_session_id=ux_session_id))
await session.execute(
statement.on_conflict_do_update(
index_elements=[
UserConsent.user_id,
UserConsent.consent_type,
UserConsent.document_version,
],
set_={
"ux_session_id": statement.excluded.ux_session_id,
"client_ip": func.coalesce(
UserConsent.client_ip, statement.excluded.client_ip
),
"user_agent_hash": func.coalesce(
UserConsent.user_agent_hash,
statement.excluded.user_agent_hash,
),
"device_json": case(
(UserConsent.device_json == {}, statement.excluded.device_json),
else_=UserConsent.device_json,
),
},
)
)
session.add(
audit(
"consent.recorded",
context,
user.id,
metadata={"consent_versions": versions},
)
)
await session.commit()
return {"recorded_at": now, "versions": versions}
async def start_session(
session: AsyncSession, user: UserIdentity, body: SessionStartRequest, request_id: str
session: AsyncSession, user: UserIdentity, body: SessionStartRequest, context: AuditContext
) -> dict[str, Any]:
now, session_id = datetime.now(UTC), uuid.uuid4()
device_hash = (
hashlib.sha256(body.device.device_id.encode()).hexdigest()
if body.device.device_id
else None
)
device = device_snapshot(body.device)
session.add(
UxSession(
id=session_id,
@@ -273,11 +366,22 @@ async def start_session(
start_reason=body.start_reason,
platform=body.device.platform,
app_version=body.device.app_version,
device_id_hash=device_hash,
device_id_hash=device["device_id_hash"],
started_at=now,
)
)
session.add(audit("session_start", request_id, user.id, ux_session_id=session_id))
session.add(
audit(
"session_start",
context.with_ux_session(session_id),
user.id,
metadata={
"start_reason": body.start_reason,
"platform": body.device.platform,
"app_version": body.device.app_version,
},
)
)
await session.commit()
return {"ux_session_id": session_id, "started_at": now}
@@ -407,7 +511,7 @@ async def owned_dialog(session: AsyncSession, user_id: uuid.UUID, dialog_id: uui
async def create_dialog(
session: AsyncSession, user: UserIdentity, request_id: str, idempotency_key: str
session: AsyncSession, user: UserIdentity, context: AuditContext, idempotency_key: str
) -> tuple[dict[str, Any], int]:
scope = "dialogs.create"
fingerprint = hashlib.sha256(f"POST:/api/v1/dialogs:{user.id}".encode()).hexdigest()
@@ -454,7 +558,16 @@ async def create_dialog(
return result, 200
dialog = Dialog(id=uuid.uuid4(), user_id=user.id, status="open")
session.add(dialog)
session.add(audit("dialog.created", request_id, user.id, "dialog", dialog.id))
session.add(
audit(
"dialog.created",
context,
user.id,
"dialog",
dialog.id,
metadata={"status": dialog.status},
)
)
result = dialog_dto(dialog)
session.add(
IdempotencyRecord(
@@ -482,7 +595,7 @@ async def init_attachment(
body: AttachmentInitRequest,
snapshot: SettingsSnapshot,
s3: S3Client,
request_id: str,
context: AuditContext,
) -> dict[str, Any]:
await owned_dialog(session, user.id, dialog_id)
extension = PurePath(body.file_name).suffix.lower().lstrip(".")
@@ -514,7 +627,14 @@ async def init_attachment(
)
session.add(item)
session.add(
audit("attachment.upload_initialized", request_id, user.id, "attachment", attachment_id)
audit(
"attachment.upload_initialized",
context,
user.id,
"attachment",
attachment_id,
metadata={"mime_type": body.mime_type, "size_bytes": body.size_bytes},
)
)
await session.commit()
url = await s3.presign_put(key, body.mime_type, ttl)
@@ -533,7 +653,7 @@ async def complete_attachment(
attachment_id: uuid.UUID,
body: AttachmentCompleteRequest,
s3: S3Client,
request_id: str,
context: AuditContext,
) -> dict[str, Any]:
item = (
await session.execute(
@@ -560,7 +680,16 @@ async def complete_attachment(
raise DomainError("attachment_checksum_mismatch", 400, "Uploaded metadata does not match")
item.checksum_sha256 = checksum
item.completed_at = datetime.now(UTC)
session.add(audit("attachment.upload_completed", request_id, user.id, "attachment", item.id))
session.add(
audit(
"attachment.upload_completed",
context,
user.id,
"attachment",
item.id,
metadata={"mime_type": item.mime_type, "size_bytes": item.size_bytes},
)
)
await session.commit()
return attachment_dto(item)
@@ -583,7 +712,7 @@ async def send_message(
dialog_id: uuid.UUID,
body: MessageRequest,
idem_key: str,
request_id: str,
context: AuditContext,
settings: Settings,
safety: SafetyClient,
openlines: OpenLinesClient,
@@ -672,7 +801,16 @@ async def send_message(
session.add(message)
if attachment:
attachment.message_id = message_id
session.add(audit("message.submitted", request_id, user.id, "message", message_id))
session.add(
audit(
"message.submitted",
context,
user.id,
"message",
message_id,
metadata={"content_kind": body.content_kind, "delivery_status": "processing"},
)
)
await session.commit()
await publish_message(fanout, message, settings, [attachment] if attachment else [])
payload: dict[str, Any] = {"message_id": str(message_id), "content_kind": kind, "text": text}
@@ -684,8 +822,9 @@ async def send_message(
"mime_type": attachment.mime_type,
"size_bytes": attachment.size_bytes,
}
outbox: DeliveryOutbox | None = None
try:
verdict = await safety.check(payload, request_id)
verdict = await safety.check(payload, context.request_id)
if verdict["_status"] == 203:
task_id = verdict["task_id"]
task = SafetyTask(
@@ -703,7 +842,7 @@ async def send_message(
deadline = time_monotonic() + settings.message_safety_task_poll_max_sec
while time_monotonic() < deadline:
await sleep(settings.message_safety_task_poll_interval_sec)
verdict = await safety.poll(task_id, request_id)
verdict = await safety.poll(task_id, context.request_id)
if verdict["_status"] != 203:
break
else:
@@ -719,7 +858,20 @@ async def send_message(
if attachment and attachment.quarantine_object_key:
attachment.scan_status = "infected"
await s3.delete_quarantine(attachment.quarantine_object_key)
session.add(audit("message.blocked", request_id, user.id, "message", message.id))
session.add(
audit(
"message.blocked",
context,
user.id,
"message",
message.id,
metadata={
"content_kind": message.content_kind,
"safety_status": message.safety_status,
"delivery_status": message.delivery_status,
},
)
)
idem.status = "completed"
idem.response_status = 422
idem.response_body_json = {
@@ -765,18 +917,35 @@ async def send_message(
),
},
},
next_attempt_at=datetime.now(UTC),
# Keep the row recoverable after a process crash, but do not let the
# delivery worker race the synchronous first attempt.
next_attempt_at=datetime.now(UTC)
+ timedelta(seconds=settings.bitrix_local_app_http_timeout_sec + 5),
)
session.add(outbox)
await session.commit()
await publish_message_status(fanout, message, settings)
delivery_payload = await fresh_openlines_payload(outbox.payload_json, s3)
await openlines.send(message.id, delivery_payload, request_id)
await openlines.send(message.id, delivery_payload, context.request_id)
message.delivery_status = "delivered"
dialog.status = "waiting_for_company"
dialog.last_message_at = datetime.now(UTC)
outbox.status = "delivered"
session.add(audit("message.delivered", request_id, user.id, "message", message.id))
session.add(
audit(
"message.delivered",
context,
user.id,
"message",
message.id,
metadata={
"content_kind": message.content_kind,
"safety_status": message.safety_status,
"delivery_status": message.delivery_status,
},
outcome="blocked",
)
)
result = message_dto(message, [attachment] if attachment else [])
idem.status = "completed"
idem.response_status = 201
@@ -789,7 +958,31 @@ async def send_message(
raise
except DependencyFailure as exc:
message.delivery_status = "failed"
session.add(audit("message.failed", request_id, user.id, "message", message.id))
if outbox is not None:
outbox.attempt_count += 1
outbox.status = "retry"
outbox.next_attempt_at = datetime.now(UTC) + timedelta(
seconds=min(3600, 2**outbox.attempt_count)
)
outbox.last_error_code = (
"dependency_timeout" if exc.timeout else "dependency_unavailable"
)
outbox.locked_at = None
outbox.locked_by = None
session.add(
audit(
"message.failed",
context,
user.id,
"message",
message.id,
metadata={
"content_kind": message.content_kind,
"delivery_status": message.delivery_status,
},
outcome="failed",
)
)
await session.commit()
await publish_message_status(fanout, message, settings)
raise DomainError(
@@ -802,7 +995,7 @@ async def send_message(
async def apply_inbox(
session: AsyncSession,
event: OpenLinesInbox,
request_id: str,
context: AuditContext,
snapshot: SettingsSnapshot,
s3: S3Client,
http: httpx.AsyncClient,
@@ -910,7 +1103,16 @@ async def apply_inbox(
dialog.status = "waiting_for_client"
dialog.last_message_at = event.occurred_at
receipt.status = "applied"
session.add(audit("openlines.inbox_applied", request_id, dialog.user_id, "dialog", dialog.id))
session.add(
audit(
"openlines.inbox_applied",
context,
None,
"dialog",
dialog.id,
metadata={"event_type": event.event_type},
)
)
await session.commit()
if message is not None:
await publish_message(fanout, message, settings, [attachment] if attachment else [])