From ff36b9f3344b4d6688c7de21e700ff588fd66d8f Mon Sep 17 00:00:00 2001
From: Kris <2893855659@qq.com>
Date: Sun, 31 May 2026 21:42:35 +0800
Subject: [PATCH] =?UTF-8?q?feat(channel-domain):=20=E6=96=B0=E5=A2=9E?=
 =?UTF-8?q?=E9=A2=86=E5=9F=9F=E6=A8=A1=E5=9E=8B=E3=80=81=E5=BC=82=E5=B8=B8?=
 =?UTF-8?q?=E3=80=81=E7=AB=AF=E5=8F=A3=E5=8F=8A=E5=9F=BA=E7=A1=80=E5=B7=A5?=
 =?UTF-8?q?=E5=85=B7=E7=B1=BB?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

新增消息日志、发件箱、会话等领域模型，新增多类型业务异常定义，添加工厂协议端口与数据库工作单元实现，同时补充飞书、钩子、Web渠道的请求验证器，以及数据转换类和异常翻译工具
---
 .../yuxi/channel/channels/feishu/verifier.py  | 66 +++++++++++++++++++
 .../yuxi/channel/channels/hooks/verifier.py   | 42 ++++++++++++
 .../yuxi/channel/channels/web/verifier.py     | 39 +++++++++++
 .../channel/domain/event/message_dropped.py   | 14 ++++
 .../channel/domain/exception/cache_error.py   |  7 ++
 .../domain/exception/concurrency_error.py     |  7 ++
 .../domain/exception/duplicate_binding.py     | 11 ++++
 .../exception/duplicate_entity_error.py       |  7 ++
 .../exception/entity_not_found_error.py       |  7 ++
 .../domain/exception/invalid_agent_config.py  |  9 +++
 .../exception/session_not_found_error.py      |  9 +++
 .../model/message/channel_message_data.py     | 13 ++++
 .../domain/model/message_log/__init__.py      |  0
 .../model/message_log/message_log_data.py     | 35 ++++++++++
 .../domain/model/outbox/outbox_status.py      | 12 ++++
 .../port/external/authentication_port.py      |  8 +++
 .../external/channel_request_verifier_port.py | 22 +++++++
 .../domain/port/internal/unit_of_work.py      | 15 +++++
 .../converter/message_log_converter.py        | 25 +++++++
 .../persistence/converter/outbox_converter.py | 21 ++++++
 .../converter/session_converter.py            | 17 +++++
 .../persistence/exception_translator.py       | 22 +++++++
 .../channel/infrastructure/persistence/uow.py | 17 +++++
 23 files changed, 425 insertions(+)
 create mode 100644 backend/package/yuxi/channel/channels/feishu/verifier.py
 create mode 100644 backend/package/yuxi/channel/channels/hooks/verifier.py
 create mode 100644 backend/package/yuxi/channel/channels/web/verifier.py
 create mode 100644 backend/package/yuxi/channel/domain/event/message_dropped.py
 create mode 100644 backend/package/yuxi/channel/domain/exception/cache_error.py
 create mode 100644 backend/package/yuxi/channel/domain/exception/concurrency_error.py
 create mode 100644 backend/package/yuxi/channel/domain/exception/duplicate_binding.py
 create mode 100644 backend/package/yuxi/channel/domain/exception/duplicate_entity_error.py
 create mode 100644 backend/package/yuxi/channel/domain/exception/entity_not_found_error.py
 create mode 100644 backend/package/yuxi/channel/domain/exception/invalid_agent_config.py
 create mode 100644 backend/package/yuxi/channel/domain/exception/session_not_found_error.py
 create mode 100644 backend/package/yuxi/channel/domain/model/message/channel_message_data.py
 create mode 100644 backend/package/yuxi/channel/domain/model/message_log/__init__.py
 create mode 100644 backend/package/yuxi/channel/domain/model/message_log/message_log_data.py
 create mode 100644 backend/package/yuxi/channel/domain/model/outbox/outbox_status.py
 create mode 100644 backend/package/yuxi/channel/domain/port/external/authentication_port.py
 create mode 100644 backend/package/yuxi/channel/domain/port/external/channel_request_verifier_port.py
 create mode 100644 backend/package/yuxi/channel/domain/port/internal/unit_of_work.py
 create mode 100644 backend/package/yuxi/channel/infrastructure/persistence/converter/message_log_converter.py
 create mode 100644 backend/package/yuxi/channel/infrastructure/persistence/converter/outbox_converter.py
 create mode 100644 backend/package/yuxi/channel/infrastructure/persistence/converter/session_converter.py
 create mode 100644 backend/package/yuxi/channel/infrastructure/persistence/exception_translator.py
 create mode 100644 backend/package/yuxi/channel/infrastructure/persistence/uow.py

diff --git a/backend/package/yuxi/channel/channels/feishu/verifier.py b/backend/package/yuxi/channel/channels/feishu/verifier.py
new file mode 100644
index 00000000..103429a1
--- /dev/null
+++ b/backend/package/yuxi/channel/channels/feishu/verifier.py
@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import logging
+import time
+
+from yuxi.channel.domain.port.external.channel_request_verifier_port import (
+    VerifyResult,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class FeishuRequestVerifier:
+    def __init__(
+        self,
+        verification_token: str = "",
+        encrypt_key: str = "",
+        timestamp_tolerance: int = 300,
+    ):
+        self._verification_token = verification_token
+        self._encrypt_key = encrypt_key
+        self._timestamp_tolerance = timestamp_tolerance
+
+    @property
+    def channel_type(self) -> str:
+        return "feishu"
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self._encrypt_key)
+
+    @property
+    def has_verification_token(self) -> bool:
+        return bool(self._verification_token)
+
+    async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult:
+        if not self._encrypt_key:
+            return VerifyResult(passed=True, method="feishu_none", reason="no encrypt_key configured")
+
+        signature = headers.get("x-lark-signature", "")
+        timestamp = headers.get("x-lark-request-timestamp", "")
+        nonce = headers.get("x-lark-request-nonce", "")
+
+        if not signature or not timestamp:
+            return VerifyResult(passed=False, method="feishu_signature", reason="missing signature headers")
+
+        try:
+            ts = float(timestamp)
+            if abs(time.time() - ts) > self._timestamp_tolerance:
+                return VerifyResult(passed=False, method="feishu_signature", reason="timestamp expired")
+        except (ValueError, TypeError):
+            return VerifyResult(passed=False, method="feishu_signature", reason="invalid timestamp")
+
+        sign_string = f"{timestamp}{nonce}{self._encrypt_key}{body.decode('utf-8', errors='replace')}"
+        expected = hashlib.sha256(sign_string.encode("utf-8")).hexdigest()
+        if not hmac.compare_digest(signature.lower(), expected.lower()):
+            return VerifyResult(passed=False, method="feishu_signature", reason="signature mismatch")
+
+        return VerifyResult(passed=True, method="feishu_signature")
+
+    def verify_token(self, token: str) -> bool:
+        if not self._verification_token:
+            return True
+        return hmac.compare_digest(token, self._verification_token)
diff --git a/backend/package/yuxi/channel/channels/hooks/verifier.py b/backend/package/yuxi/channel/channels/hooks/verifier.py
new file mode 100644
index 00000000..6bbd7166
--- /dev/null
+++ b/backend/package/yuxi/channel/channels/hooks/verifier.py
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+
+from yuxi.channel.domain.port.external.channel_request_verifier_port import (
+    VerifyResult,
+)
+
+
+class HooksRequestVerifier:
+    @property
+    def channel_type(self) -> str:
+        return "hooks"
+
+    @property
+    def enabled(self) -> bool:
+        return True
+
+    async def verify(self, body: bytes, headers: dict[str, str], *, secret: str | None = None) -> VerifyResult:
+        if secret is None:
+            return VerifyResult(passed=False, method="hooks_none", reason="secret not configured, request rejected")
+
+        if not secret:
+            return VerifyResult(passed=True, method="hooks_none", reason="no secret configured")
+
+        auth_header = headers.get("authorization", "")
+        if auth_header.startswith("Bearer "):
+            token = auth_header[7:]
+            if hmac.compare_digest(token, secret):
+                return VerifyResult(passed=True, method="hooks_bearer")
+            return VerifyResult(passed=False, method="hooks_bearer", reason="invalid bearer token")
+
+        signature = headers.get("x-signature-256") or headers.get("x-hub-signature-256")
+        if signature:
+            sig = signature[7:] if signature.startswith("sha256=") else signature
+            expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+            if hmac.compare_digest(sig, expected):
+                return VerifyResult(passed=True, method="hooks_hmac")
+            return VerifyResult(passed=False, method="hooks_hmac", reason="invalid signature")
+
+        return VerifyResult(passed=False, method="hooks_none", reason="authentication required")
diff --git a/backend/package/yuxi/channel/channels/web/verifier.py b/backend/package/yuxi/channel/channels/web/verifier.py
new file mode 100644
index 00000000..30f75d5b
--- /dev/null
+++ b/backend/package/yuxi/channel/channels/web/verifier.py
@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.port.external.channel_request_verifier_port import (
+    VerifyResult,
+)
+from yuxi.channel.domain.port.external.authentication_port import (
+    AuthenticationPort,
+)
+
+
+class WebRequestVerifier:
+    def __init__(self, auth_service: AuthenticationPort | None = None, *, allow_anonymous: bool = False):
+        self._auth_service = auth_service
+        self._allow_anonymous = allow_anonymous
+
+    @property
+    def channel_type(self) -> str:
+        return "web"
+
+    @property
+    def enabled(self) -> bool:
+        return self._auth_service is not None
+
+    async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult:
+        auth_header = headers.get("authorization", "")
+        if not auth_header:
+            if self._allow_anonymous:
+                return VerifyResult(passed=True, method="web_none", reason="no auth header, anonymous allowed")
+            if not self._auth_service:
+                return VerifyResult(passed=True, method="web_none", reason="no auth_service configured")
+            return VerifyResult(passed=False, method="web_bearer", reason="no credentials configured")
+
+        if not self._auth_service:
+            return VerifyResult(passed=True, method="web_none", reason="no auth_service configured")
+
+        passed, reason = await self._auth_service.authenticate(auth_header)
+        if passed:
+            return VerifyResult(passed=True, method="web_bearer")
+        return VerifyResult(passed=False, method="web_bearer", reason=reason)
diff --git a/backend/package/yuxi/channel/domain/event/message_dropped.py b/backend/package/yuxi/channel/domain/event/message_dropped.py
new file mode 100644
index 00000000..dd2c2ad7
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/event/message_dropped.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class MessageDropped:
+    message_id: str
+    channel_type: str
+    reason: str
+    session_id: str = ""
+    trace_id: str = ""
+    metadata: dict[str, Any] = field(default_factory=dict)
diff --git a/backend/package/yuxi/channel/domain/exception/cache_error.py b/backend/package/yuxi/channel/domain/exception/cache_error.py
new file mode 100644
index 00000000..ed522aaf
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/exception/cache_error.py
@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.recoverable_error import RecoverableError
+
+
+class CacheError(RecoverableError):
+    pass
diff --git a/backend/package/yuxi/channel/domain/exception/concurrency_error.py b/backend/package/yuxi/channel/domain/exception/concurrency_error.py
new file mode 100644
index 00000000..26562407
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/exception/concurrency_error.py
@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.unrecoverable_error import UnrecoverableError
+
+
+class ConcurrencyError(UnrecoverableError):
+    pass
diff --git a/backend/package/yuxi/channel/domain/exception/duplicate_binding.py b/backend/package/yuxi/channel/domain/exception/duplicate_binding.py
new file mode 100644
index 00000000..4919ad2f
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/exception/duplicate_binding.py
@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.channel_error import ChannelError
+
+
+class DuplicateBindingException(ChannelError):
+    def __init__(self, channel_type: str, account_id: str, group_id: str):
+        self.channel_type = channel_type
+        self.account_id = account_id
+        self.group_id = group_id
+        super().__init__(f"duplicate binding: {channel_type}/{account_id}/{group_id}")
diff --git a/backend/package/yuxi/channel/domain/exception/duplicate_entity_error.py b/backend/package/yuxi/channel/domain/exception/duplicate_entity_error.py
new file mode 100644
index 00000000..8ea33c20
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/exception/duplicate_entity_error.py
@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.recoverable_error import RecoverableError
+
+
+class DuplicateEntityError(RecoverableError):
+    pass
diff --git a/backend/package/yuxi/channel/domain/exception/entity_not_found_error.py b/backend/package/yuxi/channel/domain/exception/entity_not_found_error.py
new file mode 100644
index 00000000..3399c6d0
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/exception/entity_not_found_error.py
@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.unrecoverable_error import UnrecoverableError
+
+
+class EntityNotFoundError(UnrecoverableError):
+    pass
diff --git a/backend/package/yuxi/channel/domain/exception/invalid_agent_config.py b/backend/package/yuxi/channel/domain/exception/invalid_agent_config.py
new file mode 100644
index 00000000..2a495bee
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/exception/invalid_agent_config.py
@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.channel_error import ChannelError
+
+
+class InvalidAgentConfigException(ChannelError):
+    def __init__(self, agent_config_id: int):
+        self.agent_config_id = agent_config_id
+        super().__init__(f"agent_config_id {agent_config_id} does not exist")
diff --git a/backend/package/yuxi/channel/domain/exception/session_not_found_error.py b/backend/package/yuxi/channel/domain/exception/session_not_found_error.py
new file mode 100644
index 00000000..ef2a02ba
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/exception/session_not_found_error.py
@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.channel_error import ChannelError
+
+
+class SessionNotFoundError(ChannelError):
+    def __init__(self, *, thread_id: str):
+        self.thread_id = thread_id
+        super().__init__(f"Session not found for thread_id: {thread_id}")
diff --git a/backend/package/yuxi/channel/domain/model/message/channel_message_data.py b/backend/package/yuxi/channel/domain/model/message/channel_message_data.py
new file mode 100644
index 00000000..6aea83d8
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/model/message/channel_message_data.py
@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class ChannelMessageData:
+    id: int
+    thread_id: str
+    role: str
+    content: str
+    message_id: str | None
+    extra_metadata: dict | None
diff --git a/backend/package/yuxi/channel/domain/model/message_log/__init__.py b/backend/package/yuxi/channel/domain/model/message_log/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/backend/package/yuxi/channel/domain/model/message_log/message_log_data.py b/backend/package/yuxi/channel/domain/model/message_log/message_log_data.py
new file mode 100644
index 00000000..78fdb761
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/model/message_log/message_log_data.py
@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class MessageLogData:
+    id: int
+    trace_id: str
+    message_id: str
+    channel_type: str
+    conversation_id: int | None
+    session_id: str | None
+    direction: str
+    sender_id: str | None
+    content_summary: str | None
+    agent_config_id: int | None
+    status: str
+    pipeline_result: str | None
+    abort_reason: str | None
+    worker_result: str | None
+    error_message: str | None
+    processing_time_ms: int | None
+
+
+@dataclass
+class MessageLogQuery:
+    channel_type: str | None = None
+    conversation_id: int | None = None
+    status: str | None = None
+    direction: str | None = None
+    pipeline_result: str | None = None
+    worker_result: str | None = None
+    limit: int = 50
+    offset: int = 0
diff --git a/backend/package/yuxi/channel/domain/model/outbox/outbox_status.py b/backend/package/yuxi/channel/domain/model/outbox/outbox_status.py
new file mode 100644
index 00000000..9a472d6a
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/model/outbox/outbox_status.py
@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from enum import StrEnum
+
+
+class OutboxStatus(StrEnum):
+    PENDING = "pending"
+    PROCESSING = "processing"
+    RETRYING = "retrying"
+    SENT = "sent"
+    DEAD = "dead"
+    FAILED = "failed"
diff --git a/backend/package/yuxi/channel/domain/port/external/authentication_port.py b/backend/package/yuxi/channel/domain/port/external/authentication_port.py
new file mode 100644
index 00000000..704d23a2
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/port/external/authentication_port.py
@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class AuthenticationPort(Protocol):
+    async def authenticate(self, auth_header: str, *, client_id: str = "unknown") -> tuple[bool, str]: ...
diff --git a/backend/package/yuxi/channel/domain/port/external/channel_request_verifier_port.py b/backend/package/yuxi/channel/domain/port/external/channel_request_verifier_port.py
new file mode 100644
index 00000000..41533233
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/port/external/channel_request_verifier_port.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Protocol, runtime_checkable
+
+
+@dataclass(frozen=True)
+class VerifyResult:
+    passed: bool
+    method: str = ""
+    reason: str = ""
+
+
+@runtime_checkable
+class ChannelRequestVerifierPort(Protocol):
+    @property
+    def channel_type(self) -> str: ...
+
+    @property
+    def enabled(self) -> bool: ...
+
+    async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult: ...
diff --git a/backend/package/yuxi/channel/domain/port/internal/unit_of_work.py b/backend/package/yuxi/channel/domain/port/internal/unit_of_work.py
new file mode 100644
index 00000000..41a74a18
--- /dev/null
+++ b/backend/package/yuxi/channel/domain/port/internal/unit_of_work.py
@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class UnitOfWork(Protocol):
+    async def commit(self) -> None: ...
+
+    async def rollback(self) -> None: ...
+
+
+@runtime_checkable
+class HasSession(Protocol):
+    def session(self) -> object: ...
diff --git a/backend/package/yuxi/channel/infrastructure/persistence/converter/message_log_converter.py b/backend/package/yuxi/channel/infrastructure/persistence/converter/message_log_converter.py
new file mode 100644
index 00000000..45ccc6d9
--- /dev/null
+++ b/backend/package/yuxi/channel/infrastructure/persistence/converter/message_log_converter.py
@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.model.message_log.message_log_data import MessageLogData
+from yuxi.storage.postgres.models_channel import ChannelMessageLog
+
+
+def to_data(row: ChannelMessageLog) -> MessageLogData:
+    return MessageLogData(
+        id=row.id,
+        trace_id=row.trace_id,
+        message_id=row.message_id,
+        channel_type=row.channel_type,
+        conversation_id=row.conversation_id,
+        session_id=row.session_id,
+        direction=row.direction,
+        sender_id=row.sender_id,
+        content_summary=row.content_summary,
+        agent_config_id=row.agent_config_id,
+        status=row.status,
+        pipeline_result=row.pipeline_result,
+        abort_reason=row.abort_reason,
+        worker_result=row.worker_result,
+        error_message=row.error_message,
+        processing_time_ms=row.processing_time_ms,
+    )
diff --git a/backend/package/yuxi/channel/infrastructure/persistence/converter/outbox_converter.py b/backend/package/yuxi/channel/infrastructure/persistence/converter/outbox_converter.py
new file mode 100644
index 00000000..4d470725
--- /dev/null
+++ b/backend/package/yuxi/channel/infrastructure/persistence/converter/outbox_converter.py
@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.model.outbox.outbox_entry import OutboxEntry
+from yuxi.storage.postgres.models_channel import ChannelOutbox
+
+
+def to_domain(row: ChannelOutbox) -> OutboxEntry:
+    return OutboxEntry(
+        id=row.id,
+        message_id=row.message_id,
+        session_id=row.session_id,
+        channel_type=row.channel_type,
+        content=row.content,
+        status=row.status,
+        retry_count=row.retry_count,
+        max_retries=row.max_retries,
+        next_retry_at=row.next_retry_at,
+        last_error=row.last_error,
+        trace_id=row.trace_id,
+        extra_metadata=row.extra_metadata,
+    )
diff --git a/backend/package/yuxi/channel/infrastructure/persistence/converter/session_converter.py b/backend/package/yuxi/channel/infrastructure/persistence/converter/session_converter.py
new file mode 100644
index 00000000..7d33ce73
--- /dev/null
+++ b/backend/package/yuxi/channel/infrastructure/persistence/converter/session_converter.py
@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.model.session.channel_session import ChannelSession
+from yuxi.storage.postgres.models_business import Conversation
+
+
+def to_domain(row: Conversation) -> ChannelSession:
+    return ChannelSession(
+        id=row.id,
+        thread_id=row.thread_id,
+        user_id=row.user_id,
+        agent_id=row.agent_id,
+        channel_type=row.channel_type,
+        channel_session_key=row.channel_session_key,
+        status=row.status,
+        title=row.title,
+    )
diff --git a/backend/package/yuxi/channel/infrastructure/persistence/exception_translator.py b/backend/package/yuxi/channel/infrastructure/persistence/exception_translator.py
new file mode 100644
index 00000000..c24f04dc
--- /dev/null
+++ b/backend/package/yuxi/channel/infrastructure/persistence/exception_translator.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from sqlalchemy.exc import IntegrityError as SaIntegrityError
+from sqlalchemy.exc import OperationalError as SaOperationalError
+
+from yuxi.channel.domain.exception.concurrency_error import ConcurrencyError
+from yuxi.channel.domain.exception.duplicate_entity_error import DuplicateEntityError
+from yuxi.channel.domain.exception.entity_not_found_error import EntityNotFoundError
+from yuxi.channel.domain.exception.unrecoverable_error import UnrecoverableError
+
+
+def translate_db_exception(exc: Exception) -> UnrecoverableError:
+    if isinstance(exc, SaIntegrityError):
+        msg = str(exc.orig) if hasattr(exc, "orig") else str(exc)
+        if "unique" in msg.lower() or "duplicate" in msg.lower():
+            return DuplicateEntityError(msg)
+        if "foreign key" in msg.lower():
+            return EntityNotFoundError(msg)
+        return ConcurrencyError(msg)
+    if isinstance(exc, SaOperationalError):
+        return UnrecoverableError(f"Database operational error: {exc}")
+    return UnrecoverableError(str(exc))
diff --git a/backend/package/yuxi/channel/infrastructure/persistence/uow.py b/backend/package/yuxi/channel/infrastructure/persistence/uow.py
new file mode 100644
index 00000000..044910e3
--- /dev/null
+++ b/backend/package/yuxi/channel/infrastructure/persistence/uow.py
@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+
+class SqlAlchemyUnitOfWork:
+    def __init__(self, session: AsyncSession):
+        self._session = session
+
+    async def commit(self) -> None:
+        await self._session.commit()
+
+    async def rollback(self) -> None:
+        await self._session.rollback()
+
+    def session(self) -> AsyncSession:
+        return self._session