diff --git a/app/db/oper/site.py b/app/db/oper/site.py index 559df23d4..4733cb784 100644 --- a/app/db/oper/site.py +++ b/app/db/oper/site.py @@ -2,6 +2,7 @@ from datetime import datetime from typing import Any, List, Mapping, Tuple, Optional from sqlalchemy import delete as sqlalchemy_delete +from sqlalchemy.orm import Session from app.db.base import DbOper from app.db.models.site import Site @@ -285,79 +286,90 @@ class SiteOper(DbOper): 更新站点图标 """ icon_base64 = f"data:image/ico;base64,{icon_base64}" if icon_base64 else "" - siteicon = self.get_icon_by_domain(domain) - if not siteicon: - self._stage_create( - SiteIcon(name=name, domain=domain, url=icon_url, base64=icon_base64) - ) - elif icon_base64: - self._stage_update(siteicon, { - "url": icon_url, - "base64": icon_base64 - }) + + def write(db: Session) -> None: + """在同一同步事务中查询并更新站点图标。""" + siteicon = SiteIcon.get_by_domain(db, domain) + if not siteicon: + db.add(SiteIcon( + name=name, + domain=domain, + url=icon_url, + base64=icon_base64, + )) + elif icon_base64: + siteicon.url = icon_url + siteicon.base64 = icon_base64 + + self._execute_sync_write(write) return True def success(self, domain: str, seconds: Optional[int] = None): """ 站点访问成功 """ - lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - sta = SiteStatistic.get_by_domain(self._db, domain) - if sta: - # 使用深复制确保 note 是全新的字典对象 - note = dict(sta.note) if sta.note else {} - avg_seconds = None + def write(db: Session) -> None: + """在同一同步事务中读取并更新站点统计。""" + lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + sta = SiteStatistic.get_by_domain(db, domain) + if sta: + # 使用深复制确保 note 是全新的字典对象 + note = dict(sta.note) if sta.note else {} + avg_seconds = None - if seconds is not None: - note[lst_date] = seconds or 1 - avg_times = len(note.keys()) - if avg_times > 10: - note = dict(sorted(note.items(), key=lambda x: x[0], reverse=True)[:10]) - avg_seconds = sum([v for v in note.values()]) // avg_times + if seconds is not None: + note[lst_date] = seconds or 1 + avg_times = len(note.keys()) + if avg_times > 10: + note = dict(sorted(note.items(), key=lambda x: x[0], reverse=True)[:10]) + avg_seconds = sum([v for v in note.values()]) // avg_times - self._stage_update(sta, { - "success": sta.success + 1, - "seconds": avg_seconds or sta.seconds, - "lst_state": 0, - "lst_mod_date": lst_date, - "note": note - }) - else: - note = {} - if seconds is not None: - note = { - lst_date: seconds or 1 - } - self._stage_create(SiteStatistic( - domain=domain, - success=1, - fail=0, - seconds=seconds or 1, - lst_state=0, - lst_mod_date=lst_date, - note=note - )) + for key, value in { + "success": sta.success + 1, + "seconds": avg_seconds or sta.seconds, + "lst_state": 0, + "lst_mod_date": lst_date, + "note": note, + }.items(): + setattr(sta, key, value) + else: + note = {} + if seconds is not None: + note = {lst_date: seconds or 1} + db.add(SiteStatistic( + domain=domain, + success=1, + fail=0, + seconds=seconds or 1, + lst_state=0, + lst_mod_date=lst_date, + note=note, + )) + + self._execute_sync_write(write) def fail(self, domain: str): """ 站点访问失败 """ - lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - sta = SiteStatistic.get_by_domain(self._db, domain) - if sta: - self._stage_update(sta, { - "fail": sta.fail + 1, - "lst_state": 1, - "lst_mod_date": lst_date - }) - else: - self._stage_create(SiteStatistic( - domain=domain, - success=0, - fail=1, - lst_state=1, - lst_mod_date=lst_date - )) + def write(db: Session) -> None: + """在同一同步事务中读取并更新站点失败统计。""" + lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + sta = SiteStatistic.get_by_domain(db, domain) + if sta: + sta.fail += 1 + sta.lst_state = 1 + sta.lst_mod_date = lst_date + else: + db.add(SiteStatistic( + domain=domain, + success=0, + fail=1, + lst_state=1, + lst_mod_date=lst_date, + )) + + self._execute_sync_write(write) async def async_success(self, domain: str, seconds: Optional[int] = None): """ diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index 04cff61f1..993bbdae8 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -130,6 +130,7 @@ from app.startup.subscription import ( ) from app.startup.chain_events import TransactionalChainDurableEventWriter from app.startup.download_failure import TransactionalDownloadFailureRepository +from app.startup.site import TransactionalSiteRepository from app.startup.workflow import TransactionalWorkflowExecutionService from app.startup.transaction import TransactionalWriteRunner from app.startup.context import ( @@ -685,7 +686,10 @@ async def init_modules() -> HostRuntime: workflow_execution = TransactionalWorkflowExecutionService(SessionFactory) configure_workflow_legacy_writer(workflow_execution) configure_chain_data_ports( - site=lambda: SiteOper(), + site=lambda: TransactionalSiteRepository( + sync_session=SessionFactory, + async_session=async_session_scope, + ), subscribe=lambda: SubscribeOper(), workflow=lambda: WorkflowOper(), download_history=lambda: DownloadHistoryOper(), @@ -720,13 +724,19 @@ async def init_modules() -> HostRuntime: configure_passkey_service(PasskeyService(repository=PassKeyOper())) configure_transfer_history_provider(lambda: TransferHistoryOper()) configure_site_query_service(SiteQueryService(repository=SiteOper())) - configure_site_health_service(SiteHealthService(repository=SiteOper())) + configure_site_health_service(SiteHealthService(repository=TransactionalSiteRepository( + sync_session=SessionFactory, + async_session=async_session_scope, + ))) configure_workflow_query(WorkflowQueryService(repository=WorkflowOper())) configure_agent_data_ports( agent_chat=lambda: AgentChatOper(), agent_task=lambda: AgentTaskOper(), user=lambda: UserOper(), - site=lambda: SiteOper(), + site=lambda: TransactionalSiteRepository( + sync_session=SessionFactory, + async_session=async_session_scope, + ), subscribe=lambda: SubscribeOper(), subscribe_history=lambda: SubscribeHistoryOper(), transfer_history=lambda: TransferHistoryOper(), diff --git a/app/startup/site.py b/app/startup/site.py new file mode 100644 index 000000000..8f07bb42b --- /dev/null +++ b/app/startup/site.py @@ -0,0 +1,193 @@ +"""站点 Chain 端口的显式会话与事务适配器。""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from contextlib import AbstractAsyncContextManager +from typing import Any, TypeVar + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Session + +from app.db.oper.site import SiteOper +from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork + + +T = TypeVar("T") + + +class TransactionalSiteRepository: + """为同步 Chain 站点端口和异步健康统计提供短生命周期会话。""" + + def __init__( + self, + *, + sync_session: Callable[[], Session], + async_session: Callable[[], AbstractAsyncContextManager[AsyncSession]], + ) -> None: + """保存同步会话工厂和异步会话上下文工厂。""" + self._sync_session = sync_session + self._async_session = async_session + + def _read(self, operation: Callable[[SiteOper], T]) -> T: + """在独立同步会话中执行只读站点操作。""" + with self._sync_session() as session: + return operation(SiteOper(db=session)) + + def _write(self, operation: Callable[[SiteOper], T]) -> T: + """在独立同步 UoW 中执行站点写操作。""" + with self._sync_session() as session: + session.expire_on_commit = False + unit_of_work = SqlAlchemyUnitOfWork(session) + try: + result = operation(SiteOper(db=session)) + unit_of_work.commit() + return result + except Exception: + unit_of_work.rollback() + raise + + async def _async_write( + self, + operation: Callable[[SiteOper], Awaitable[T]], + ) -> T: + """在独立异步 UoW 中执行站点写操作。""" + async with self._async_session() as session: + session.sync_session.expire_on_commit = False + unit_of_work = SqlAlchemyAsyncUnitOfWork(session) + try: + result = await operation(SiteOper(db=session)) + await unit_of_work.commit() + return result + except Exception: + await unit_of_work.rollback() + raise + + async def _async_read(self, operation: Callable[[SiteOper], Awaitable[T]]) -> T: + """在独立异步会话中执行只读站点操作。""" + async with self._async_session() as session: + return await operation(SiteOper(db=session)) + + def add(self, **kwargs: Any) -> tuple[bool, str]: + """新增站点并提交事务。""" + return self._write(lambda repository: repository.add(**kwargs)) + + def get(self, site_id: int) -> Any: + """按 ID 查询站点。""" + return self._read(lambda repository: repository.get(site_id)) + + def get_by_domain(self, domain: str) -> Any: + """按域名查询站点。""" + return self._read(lambda repository: repository.get_by_domain(domain)) + + def get_domains_by_ids(self, ids: list[int]) -> list[str | None]: + """查询一组站点 ID 对应的域名。""" + return self._read(lambda repository: repository.get_domains_by_ids(ids)) + + def list(self) -> list[Any]: + """查询全部站点。""" + return self._read(lambda repository: repository.list()) + + async def async_get(self, site_id: int) -> Any: + """异步按 ID 查询站点。""" + return await self._async_read(lambda repository: repository.async_get(site_id)) + + async def async_get_by_domain(self, domain: str) -> Any: + """异步按域名查询站点。""" + return await self._async_read( + lambda repository: repository.async_get_by_domain(domain) + ) + + async def async_get_by_name(self, name: str) -> Any: + """异步按名称查询站点。""" + return await self._async_read( + lambda repository: repository.async_get_by_name(name) + ) + + async def async_list(self) -> list[Any]: + """异步查询全部站点。""" + return await self._async_read(lambda repository: repository.async_list()) + + async def async_list_order_by_pri(self) -> list[Any]: + """异步按优先级查询站点。""" + return await self._async_read( + lambda repository: repository.async_list_order_by_pri() + ) + + async def async_update(self, site_id: int, payload: dict[str, Any]) -> Any: + """异步更新站点并提交事务。""" + return await self._async_write( + lambda repository: repository.async_update(site_id, payload) + ) + + async def async_get_userdata_by_domain( + self, + domain: str, + workdate: str | None = None, + ) -> list[Any]: + """异步查询站点用户数据。""" + return await self._async_read( + lambda repository: repository.async_get_userdata_by_domain(domain, workdate) + ) + + def update(self, site_id: int, payload: dict[str, Any]) -> Any: + """更新站点并提交事务。""" + return self._write(lambda repository: repository.update(site_id, payload)) + + def update_cookie(self, domain: str, cookies: str) -> tuple[bool, str]: + """更新站点 Cookie 并提交事务。""" + return self._write( + lambda repository: repository.update_cookie(domain, cookies) + ) + + def update_rss(self, domain: str, rss: str) -> tuple[bool, str]: + """更新站点 RSS 地址并提交事务。""" + return self._write(lambda repository: repository.update_rss(domain, rss)) + + def update_userdata( + self, + domain: str, + name: str, + payload: dict[str, Any], + ) -> tuple[bool, str]: + """更新站点用户数据并提交事务。""" + return self._write( + lambda repository: repository.update_userdata(domain, name, payload) + ) + + def update_icon( + self, + name: str, + domain: str, + icon_url: str, + icon_base64: str, + ) -> bool: + """更新站点图标并提交事务。""" + return self._write( + lambda repository: repository.update_icon( + name, + domain, + icon_url, + icon_base64, + ) + ) + + def success(self, domain: str, seconds: int | None = None) -> Any: + """记录站点访问成功并提交事务。""" + return self._write(lambda repository: repository.success(domain, seconds)) + + def fail(self, domain: str) -> Any: + """记录站点访问失败并提交事务。""" + return self._write(lambda repository: repository.fail(domain)) + + async def async_success(self, domain: str, seconds: int | None = None) -> Any: + """异步记录站点访问成功并提交事务。""" + return await self._async_write( + lambda repository: repository.async_success(domain, seconds) + ) + + async def async_fail(self, domain: str) -> Any: + """异步记录站点访问失败并提交事务。""" + return await self._async_write( + lambda repository: repository.async_fail(domain) + ) diff --git a/tests/conftest.py b/tests/conftest.py index 4092f5a7e..f05d16fef 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -148,6 +148,8 @@ def configure_plugin_system_services(): from app.db.oper.message import MessageOper from app.db.oper.passkey import PassKeyOper from app.startup.subscription import TransactionalSubscribeWriter + from app.startup.download_failure import TransactionalDownloadFailureRepository + from app.startup.site import TransactionalSiteRepository from app.startup.workflow import TransactionalWorkflowExecutionService from app.startup.transaction import TransactionalWriteRunner @@ -202,15 +204,24 @@ def configure_plugin_system_services(): ) ) + def site_repository() -> TransactionalSiteRepository: + """按生产组合根方式创建显式事务站点仓储。""" + return TransactionalSiteRepository( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + configure_chain_data_ports( - site=lambda: SiteOper(), + site=site_repository, subscribe=lambda: SubscribeOper(), workflow=lambda: WorkflowOper(), download_history=lambda: DownloadHistoryOper(), transfer_history=lambda: TransferHistoryOper(), transfer_pending=lambda: TransferPendingOper(), media_server=lambda: MediaServerOper(), - download_failure=lambda: DownloadFailureOper(), + download_failure=lambda: TransactionalDownloadFailureRepository( + SessionFactory + ), user=lambda: UserOper(), ) configure_chain_runtime_context_provider(lambda: ChainRuntimeContext( @@ -228,7 +239,7 @@ def configure_plugin_system_services(): configuration=build_chain_runtime_config(settings), )) configure_site_query_service(SiteQueryService(repository=SiteOper())) - configure_site_health_service(SiteHealthService(repository=SiteOper())) + configure_site_health_service(SiteHealthService(repository=site_repository())) configure_workflow_query(WorkflowQueryService(repository=WorkflowOper())) from app.db.oper.agenttask import AgentTaskOper from app.db.oper.plugindata import PluginDataOper @@ -236,7 +247,7 @@ def configure_plugin_system_services(): agent_chat=lambda: AgentChatOper(), agent_task=lambda: AgentTaskOper(), user=lambda: UserOper(), - site=lambda: SiteOper(), + site=site_repository, subscribe=lambda: SubscribeOper(), subscribe_history=lambda: SubscribeHistoryOper(), transfer_history=lambda: TransferHistoryOper(), diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 0d91c4603..8b33ab2f0 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6429, - "edge_sha256": "664681a500ba1c3d273568829b1b860e4dae7fe7a6c053c566583341fee7f903", + "edge_count": 6434, + "edge_sha256": "5c47cae41f5d90e1030757d0cb47db8eed373424dda5db17c900271e1d03a9c8", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -6168,6 +6168,7 @@ "app.startup.modules_initializer -> app.startup.download_failure", "app.startup.modules_initializer -> app.startup.managed_resources_initializer", "app.startup.modules_initializer -> app.startup.outbox", + "app.startup.modules_initializer -> app.startup.site", "app.startup.modules_initializer -> app.startup.subscription", "app.startup.modules_initializer -> app.startup.transaction", "app.startup.modules_initializer -> app.startup.workflow", @@ -6228,6 +6229,10 @@ "app.startup.scheduler_initializer -> app.application", "app.startup.scheduler_initializer -> app.application.scheduling", "app.startup.scheduler_initializer -> app.scheduler", + "app.startup.site -> app.db", + "app.startup.site -> app.db.oper", + "app.startup.site -> app.db.oper.site", + "app.startup.site -> app.db.uow", "app.startup.subscription -> app.adapters", "app.startup.subscription -> app.adapters.external", "app.startup.subscription -> app.adapters.external.server", @@ -6446,7 +6451,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 797, + "module_count": 798, "modules": [ "app", "app.adapters", @@ -7219,6 +7224,7 @@ "app.startup.plugins_initializer", "app.startup.routers_initializer", "app.startup.scheduler_initializer", + "app.startup.site", "app.startup.subscription", "app.startup.transaction", "app.startup.transfer_initializer", diff --git a/tests/test_db_oper_layer.py b/tests/test_db_oper_layer.py index ac982faa5..5010b4d1d 100644 --- a/tests/test_db_oper_layer.py +++ b/tests/test_db_oper_layer.py @@ -204,6 +204,22 @@ def test_site_oper_update_icon_creates_then_only_overwrites_with_content(db): assert first.startswith("data:image/ico;base64,") +def test_site_oper_icon_without_explicit_session_uses_transaction_runner(db): + """无显式会话的图标写入应由兼容事务执行器完成提交。""" + oper = SiteOper() + + oper.update_icon( + "兼容站点", + "op-icon-no-session.test", + "https://op-icon-no-session.test/favicon.ico", + "AAA", + ) + + icon = SiteIcon.get_by_domain(db.session, "op-icon-no-session.test") + assert icon.name == "兼容站点" + assert icon.base64 == "data:image/ico;base64,AAA" + + def test_site_oper_success_accumulates_and_records_state(db): """ 访问成功累加计数并把最后状态标记为成功。 @@ -219,6 +235,17 @@ def test_site_oper_success_accumulates_and_records_state(db): assert stat.seconds +def test_site_oper_statistics_without_explicit_session_use_transaction_runner(db): + """无显式会话的站点统计必须由兼容事务执行器完成提交。""" + oper = SiteOper() + + oper.success("op-stat-no-session.test", seconds=3) + oper.fail("op-stat-no-session.test") + + stat = SiteStatistic.get_by_domain(db.session, "op-stat-no-session.test") + assert (stat.success, stat.fail, stat.lst_state) == (1, 1, 1) + + def test_site_oper_success_caps_the_timing_note_at_ten_entries(db): """ 耗时记录最多保留最近 10 条，超出时丢弃最旧的。 diff --git a/tests/test_site_connectivity.py b/tests/test_site_connectivity.py new file mode 100644 index 000000000..6afa44bd7 --- /dev/null +++ b/tests/test_site_connectivity.py @@ -0,0 +1,28 @@ +"""站点连通性测试链路的数据库回归测试。""" + +from app.chain.site import SiteChain +from app.db.models.site import Site +from app.db.models.sitestatistic import SiteStatistic + + +def test_site_connectivity_records_result_without_injected_session(db, monkeypatch): + """默认站点端口完成测试后应提交统计，而不是对空会话调用 execute。""" + db.watermark(Site, SiteStatistic) + db.add(Site( + name="连通性测试站点", + domain="connectivity.test", + url="https://connectivity.test/", + is_active=True, + )) + monkeypatch.setattr( + SiteChain, + "_SiteChain__test", + lambda _self, _site: (True, "连接成功"), + ) + + status, message = SiteChain().test("https://connectivity.test/") + + statistic = SiteStatistic.get_by_domain(db.session, "connectivity.test") + assert (status, message) == (True, "连接成功") + assert statistic.success == 1 + assert statistic.lst_state == 0