From e4dcab6809c5a037580e11ac00b4458f38e1c11f Mon Sep 17 00:00:00 2001 From: damingishere-coder Date: Sun, 30 Aug 2026 13:29:49 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=94=B6=E7=B4=A7=E8=AE=A2=E5=8D=95?= =?UTF-8?q?=E4=B8=8E=E4=BB=BB=E5=8A=A1=E6=95=B0=E6=8D=AE=E4=B8=80=E8=87=B4?= =?UTF-8?q?=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/orders.py | 101 +++++- backend/app/db/preflight.py | 24 +- backend/app/db/preflight_cli.py | 18 + backend/app/db/seed.py | 16 +- backend/app/models/customer.py | 2 + backend/app/models/order.py | 8 +- backend/app/models/payment.py | 3 +- backend/app/models/task.py | 6 + backend/app/schemas/order.py | 1 + backend/app/services/dashboard.py | 2 + backend/app/services/demo_data.py | 168 ++++++++-- backend/app/services/manual_locations.py | 9 + backend/app/services/order_locations.py | 47 ++- backend/app/services/order_revisions.py | 113 +++++++ backend/app/services/orders.py | 35 +- backend/app/services/payments.py | 14 +- backend/app/services/plans.py | 10 +- backend/app/services/task_execution.py | 49 ++- .../versions/0014_consistency_guards.py | 310 ++++++++++++++++++ backend/tests/conftest.py | 50 +++ backend/tests/test_database_preflight.py | 19 ++ backend/tests/test_migrations.py | 143 ++++++++ backend/tests/test_order_api.py | 83 +++++ backend/tests/test_p18_workflow.py | 108 +++++- backend/tests/test_payment_api.py | 1 + backend/tests/test_task_api.py | 45 +++ docs/tasks/P27_DATA_CONSISTENCY_GUARDS.md | 68 ++++ .../src/features/orders/OrdersPage.test.tsx | 20 +- frontend/src/features/orders/OrdersPage.tsx | 32 +- frontend/src/features/orders/api.test.ts | 41 ++- frontend/src/features/orders/api.ts | 32 +- frontend/src/features/orders/types.ts | 1 + scripts/migrate.ps1 | 8 +- scripts/setup.ps1 | 4 +- 34 files changed, 1504 insertions(+), 87 deletions(-) create mode 100644 backend/app/db/preflight_cli.py create mode 100644 backend/app/services/order_revisions.py create mode 100644 backend/migrations/versions/0014_consistency_guards.py create mode 100644 docs/tasks/P27_DATA_CONSISTENCY_GUARDS.md diff --git a/backend/app/api/orders.py b/backend/app/api/orders.py index c8e5dcb..5ec9e8c 100644 --- a/backend/app/api/orders.py +++ b/backend/app/api/orders.py @@ -1,8 +1,9 @@ from datetime import timedelta from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, Header, HTTPException, Response, status from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, selectinload from app.db.session import get_db @@ -68,17 +69,31 @@ overpaid_amount, payment_status_for_order, require_tasks_are_rebuildable, + require_order_status_transition, replace_order_schedule, reprice_order, synchronize_task_statuses, apply_amount_adjustment, ) from app.services.payments import payment_revision, require_payment_revision +from app.services.order_revisions import ( + order_payload_hash, + order_write_revision, + reserve_order_revision, +) router = APIRouter(prefix="/api/admin/orders", tags=["admin-orders"]) DatabaseSession = Annotated[Session, Depends(get_db)] MapServicesDependency = Annotated[MapServices, Depends(get_map_services)] +OrderRevisionHeader = Annotated[ + str, + Header(alias="If-Match", pattern=r"^[0-9a-f]{64}$"), +] +IdempotencyKeyHeader = Annotated[ + str, + Header(alias="Idempotency-Key", min_length=16, max_length=128), +] def _order_load_options() -> tuple: @@ -159,6 +174,7 @@ def _order_summary(order: Order, *, customer_resolution: str | None = None) -> O ) return OrderSummary( id=order.id, + write_revision=order_write_revision(order), source_customer_id=order.customer_id, service_contact=service_contact, cat_snapshot=order.cat_snapshot or [], @@ -391,7 +407,22 @@ def create_order( payload: OrderCreate | OrderWrite, session: DatabaseSession, services: MapServicesDependency, + response: Response, + idempotency_key: IdempotencyKeyHeader, ) -> OrderDetail: + payload_hash = order_payload_hash(payload) + existing = session.scalar( + select(Order) + .options(*_order_load_options()) + .where(Order.idempotency_key == idempotency_key) + ) + if existing is not None: + if existing.idempotency_payload_hash != payload_hash: + raise HTTPException(status_code=409, detail="幂等键已用于不同的订单内容") + response.status_code = status.HTTP_200_OK + response.headers["Idempotent-Replayed"] = "true" + return _order_detail(existing) + if isinstance(payload, OrderCreate): explicit_customer_id = payload.source_customer_id or payload.customer_id selected_customer = _load_source_customer( @@ -410,8 +441,23 @@ def create_order( explicit_customer_id=explicit_customer_id, ) order = build_simple_order(payload, source_customer=resolution.customer) + order.idempotency_key = idempotency_key + order.idempotency_payload_hash = payload_hash session.add(order) - session.commit() + try: + session.commit() + except IntegrityError: + session.rollback() + replay = session.scalar( + select(Order) + .options(*_order_load_options()) + .where(Order.idempotency_key == idempotency_key) + ) + if replay is None or replay.idempotency_payload_hash != payload_hash: + raise + response.status_code = status.HTTP_200_OK + response.headers["Idempotent-Replayed"] = "true" + return _order_detail(replay) geocode_order(session, order.id, services) return _order_detail( _load_order(session, order.id), customer_resolution=resolution.result @@ -429,8 +475,23 @@ def create_order( cat_ids=payload.cat_ids, ) order = build_order(payload, cats=cats, customer=customer) + order.idempotency_key = idempotency_key + order.idempotency_payload_hash = payload_hash session.add(order) - session.commit() + try: + session.commit() + except IntegrityError: + session.rollback() + replay = session.scalar( + select(Order) + .options(*_order_load_options()) + .where(Order.idempotency_key == idempotency_key) + ) + if replay is None or replay.idempotency_payload_hash != payload_hash: + raise + response.status_code = status.HTTP_200_OK + response.headers["Idempotent-Replayed"] = "true" + return _order_detail(replay) geocode_order(session, order.id, services) return _order_detail(_load_order(session, order.id)) @@ -441,6 +502,7 @@ def patch_order( payload: OrderPatch, session: DatabaseSession, services: MapServicesDependency, + expected_revision: OrderRevisionHeader, ) -> OrderDetail: order = _load_order(session, order_id) fields = payload.model_fields_set @@ -471,6 +533,7 @@ def patch_order( requested_cat_count = payload.cat_count if payload.cat_count is not None else order.cat_count structural_change = any( ( + source_changed, requested_dates != order_schedule(order), requested_items != order.service_items, requested_cat_count != order.cat_count, @@ -495,6 +558,11 @@ def patch_order( raise HTTPException(status_code=409, detail="请刷新订单后再调整价格") require_payment_revision(order, payload.expected_financial_revision) + if source_changed and order.payments: + raise HTTPException(status_code=409, detail="订单已有收款记录，不能更换客户来源") + + reserve_order_revision(session, order, expected_revision) + if source_changed: order.customer_id = requested_customer_id order.cat_links.clear() @@ -503,7 +571,9 @@ def patch_order( if requested_customer is not None: apply_service_contact(order, customer_service_contact(requested_customer)) active_cats = [cat for cat in requested_customer.cats if cat.is_active] - order.cat_snapshot = cat_snapshot(active_cats[:requested_cat_count]) + selected_cats = active_cats[:requested_cat_count] + order.cat_snapshot = cat_snapshot(selected_cats) + order.cat_links.extend(OrderCat(cat=cat) for cat in selected_cats) if "service_contact" in fields and payload.service_contact is not None: apply_service_contact(order, payload.service_contact) elif "customer_name" in fields and payload.customer_name is not None: @@ -553,8 +623,11 @@ def retry_order_geocode( order_id: int, session: DatabaseSession, services: MapServicesDependency, + expected_revision: OrderRevisionHeader, ) -> OrderDetail: - _load_order(session, order_id) + order = _load_order(session, order_id) + reserve_order_revision(session, order, expected_revision) + session.commit() geocode_order(session, order_id, services) return _order_detail(_load_order(session, order_id)) @@ -615,6 +688,7 @@ def update_order( payload: OrderWrite, session: DatabaseSession, services: MapServicesDependency, + expected_revision: OrderRevisionHeader, ) -> OrderDetail: order = _load_order(session, order_id) existing_cat_ids = {link.cat_id for link in order.cat_links} @@ -671,6 +745,9 @@ def update_order( detail="不能在重建任务的同时把订单标记为已完成", ) + require_order_status_transition(order, payload.order_status) + reserve_order_revision(session, order, expected_revision) + order.customer_id = payload.customer_id apply_service_contact(order, customer_service_contact(customer)) order.cat_snapshot = cat_snapshot(cats) @@ -714,8 +791,11 @@ def update_order_status( order_id: int, payload: OrderStatusUpdate, session: DatabaseSession, + expected_revision: OrderRevisionHeader, ) -> OrderDetail: order = _load_order(session, order_id) + require_order_status_transition(order, payload.order_status) + reserve_order_revision(session, order, expected_revision) synchronize_task_statuses(order, payload.order_status) order.order_status = payload.order_status session.commit() @@ -723,14 +803,15 @@ def update_order_status( @router.delete("/{order_id}", status_code=status.HTTP_204_NO_CONTENT) -def delete_order(order_id: int, session: DatabaseSession) -> None: +def delete_order( + order_id: int, + session: DatabaseSession, + expected_revision: OrderRevisionHeader, +) -> None: order = _load_order(session, order_id) reason = _delete_block_reason(order) if reason is not None: raise HTTPException(status_code=409, detail=reason) + reserve_order_revision(session, order, expected_revision) session.delete(order) session.commit() - cat_snapshot, - customer_service_contact, - order_has_execution_history, - order_service_contact, diff --git a/backend/app/db/preflight.py b/backend/app/db/preflight.py index 060de50..9b1bd60 100644 --- a/backend/app/db/preflight.py +++ b/backend/app/db/preflight.py @@ -56,6 +56,20 @@ def _backup_sqlite(database_path: Path, backup_dir: Path) -> Path | None: return backup_path +def _verify_sqlite_integrity(database_path: Path | None) -> None: + if database_path is None or not database_path.exists() or database_path.stat().st_size == 0: + return + source_uri = f"file:{database_path.as_posix()}?mode=ro" + try: + with sqlite3.connect(source_uri, uri=True) as connection: + integrity = [row[0] for row in connection.execute("PRAGMA integrity_check")] + foreign_keys = list(connection.execute("PRAGMA foreign_key_check")) + except sqlite3.Error as exc: + raise DatabasePreparationError("数据库完整性检查失败，操作已停止。") from exc + if integrity != ["ok"] or foreign_keys: + raise DatabasePreparationError("数据库完整性或外键检查未通过，操作已停止。") + + def _upgrade_database(database_url: str, config_path: Path) -> None: config = Config(str(config_path)) config.set_main_option("sqlalchemy.url", database_url.replace("%", "%%")) @@ -86,16 +100,17 @@ def ensure_database_ready( """Prepare the local SQLite DB for startup, refusing unsafe external migration.""" resolved_url = get_database_url(database_url) + sqlite_path = _sqlite_database_path(resolved_url) + _verify_sqlite_integrity(sqlite_path) initial = _readiness_for(resolved_url) if initial.ready: return DatabasePreparation(False, None, initial) if initial.reason != "schema_outdated": raise DatabasePreparationError(initial.message) - sqlite_path = _sqlite_database_path(resolved_url) if sqlite_path is None: raise DatabasePreparationError( - "检测到非本机 SQLite 数据库且版本不一致；为避免误迁移，请人工运行 migrate.bat。" + "检测到非本机 SQLite 数据库且版本不一致；本地脚本不会盲目升级，请使用受控部署迁移流程。" ) try: @@ -113,5 +128,10 @@ def ensure_database_ready( if not final.ready: location = f"；备份位于 {backup_path}" if backup_path else "" raise DatabasePreparationError(f"数据库迁移后仍未就绪{location}。") + try: + _verify_sqlite_integrity(sqlite_path) + except DatabasePreparationError as exc: + location = f"；备份位于 {backup_path}" if backup_path else "" + raise DatabasePreparationError(f"{exc}{location}") from exc return DatabasePreparation(True, backup_path, final) diff --git a/backend/app/db/preflight_cli.py b/backend/app/db/preflight_cli.py new file mode 100644 index 0000000..527226f --- /dev/null +++ b/backend/app/db/preflight_cli.py @@ -0,0 +1,18 @@ +from app.db.preflight import DatabasePreparationError, ensure_database_ready + + +def main() -> None: + try: + result = ensure_database_ready() + except DatabasePreparationError as exc: + raise SystemExit(str(exc)) from exc + + if result.migrated: + backup = f"；备份：{result.backup_path}" if result.backup_path else "" + print(f"数据库迁移和完整性检查已完成{backup}。") + else: + print("数据库版本和完整性检查已通过。") + + +if __name__ == "__main__": + main() diff --git a/backend/app/db/seed.py b/backend/app/db/seed.py index d235ec5..44d8a7e 100644 --- a/backend/app/db/seed.py +++ b/backend/app/db/seed.py @@ -65,18 +65,23 @@ def _insert_demo_records(session: Session) -> None: address="仅用于开发演示，不对应任何真实地址", is_repeat_customer=False, notes=None, + seed_source=DEMO_SYSTEM_KEY, ) cat_one = Cat( name="演示猫咪一号", breed="虚构品种", personality="开发测试用虚构档案", + seed_source=DEMO_SYSTEM_KEY, ) cat_two = Cat( name="演示猫咪二号", breed="虚构品种", personality="开发测试用虚构档案", + seed_source=DEMO_SYSTEM_KEY, ) customer.cats.extend([cat_one, cat_two]) + session.add(customer) + session.flush() order = Order( customer=customer, @@ -104,6 +109,7 @@ def _insert_demo_records(session: Session) -> None: payment_status=OrderPaymentStatus.PAID, order_status=OrderStatus.CONFIRMED, notes="仅用于开发测试的虚构订单", + seed_source=DEMO_SYSTEM_KEY, ) session.add(order) session.flush() @@ -124,13 +130,14 @@ def _insert_demo_records(session: Session) -> None: sort_order=day_offset, status=TaskStatus.CONFIRMED, notes="虚构演示任务", + seed_source=DEMO_SYSTEM_KEY, ) task.items.extend( [ - TaskItem(item_type=TaskItemType.FEED), - TaskItem(item_type=TaskItemType.WATER), - TaskItem(item_type=TaskItemType.LITTER), - TaskItem(item_type=TaskItemType.PHOTO), + TaskItem(item_type=TaskItemType.FEED, seed_source=DEMO_SYSTEM_KEY), + TaskItem(item_type=TaskItemType.WATER, seed_source=DEMO_SYSTEM_KEY), + TaskItem(item_type=TaskItemType.LITTER, seed_source=DEMO_SYSTEM_KEY), + TaskItem(item_type=TaskItemType.PHOTO, seed_source=DEMO_SYSTEM_KEY), ] ) session.add(task) @@ -144,6 +151,7 @@ def _insert_demo_records(session: Session) -> None: payment_status=PaymentRecordStatus.COMPLETED, paid_at=datetime.now(timezone.utc), notes="虚构演示收款", + seed_source=DEMO_SYSTEM_KEY, ) ) diff --git a/backend/app/models/customer.py b/backend/app/models/customer.py index 9f007bb..80b8637 100644 --- a/backend/app/models/customer.py +++ b/backend/app/models/customer.py @@ -29,6 +29,7 @@ class Customer(TimestampMixin, Base): ) id: Mapped[int] = mapped_column(primary_key=True) + seed_source: Mapped[str | None] = mapped_column(String(64), index=True) system_key: Mapped[str | None] = mapped_column(String(64), unique=True, index=True) name: Mapped[str] = mapped_column(String(100), nullable=False, index=True) wechat_name: Mapped[str | None] = mapped_column(String(100)) @@ -84,6 +85,7 @@ class Cat(TimestampMixin, Base): ) id: Mapped[int] = mapped_column(primary_key=True) + seed_source: Mapped[str | None] = mapped_column(String(64), index=True) customer_id: Mapped[int] = mapped_column( ForeignKey("customers.id", ondelete="CASCADE"), nullable=False, diff --git a/backend/app/models/order.py b/backend/app/models/order.py index 7eb6037..bed2064 100644 --- a/backend/app/models/order.py +++ b/backend/app/models/order.py @@ -2,7 +2,7 @@ from decimal import Decimal from typing import TYPE_CHECKING -from sqlalchemy import Boolean, CheckConstraint, Date, ForeignKey, Index, JSON, Numeric, String, Text +from sqlalchemy import Boolean, CheckConstraint, Date, ForeignKey, Index, Integer, JSON, Numeric, String, Text from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db.base import Base @@ -54,6 +54,12 @@ class Order(TimestampMixin, Base): ) id: Mapped[int] = mapped_column(primary_key=True) + seed_source: Mapped[str | None] = mapped_column(String(64), index=True) + write_revision_number: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default="0" + ) + idempotency_key: Mapped[str | None] = mapped_column(String(128), unique=True, index=True) + idempotency_payload_hash: Mapped[str | None] = mapped_column(String(64)) customer_id: Mapped[int | None] = mapped_column( ForeignKey("customers.id", ondelete="SET NULL"), nullable=True, diff --git a/backend/app/models/payment.py b/backend/app/models/payment.py index 473c570..76959c1 100644 --- a/backend/app/models/payment.py +++ b/backend/app/models/payment.py @@ -2,7 +2,7 @@ from decimal import Decimal from typing import TYPE_CHECKING -from sqlalchemy import CheckConstraint, Date, DateTime, ForeignKey, Index, Numeric, Text +from sqlalchemy import CheckConstraint, Date, DateTime, ForeignKey, Index, Numeric, String, Text from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db.base import Base @@ -38,6 +38,7 @@ class Payment(TimestampMixin, Base): ) id: Mapped[int] = mapped_column(primary_key=True) + seed_source: Mapped[str | None] = mapped_column(String(64), index=True) order_id: Mapped[int] = mapped_column( ForeignKey("orders.id", ondelete="RESTRICT"), nullable=False, diff --git a/backend/app/models/task.py b/backend/app/models/task.py index 5fac0ca..df840cb 100644 --- a/backend/app/models/task.py +++ b/backend/app/models/task.py @@ -48,6 +48,10 @@ class Task(TimestampMixin, Base): ) id: Mapped[int] = mapped_column(primary_key=True) + seed_source: Mapped[str | None] = mapped_column(String(64), index=True) + execution_revision_number: Mapped[int] = mapped_column( + nullable=False, default=0, server_default="0" + ) order_id: Mapped[int] = mapped_column( ForeignKey("orders.id", ondelete="CASCADE"), nullable=False, @@ -103,6 +107,7 @@ class TaskItem(Base): ) id: Mapped[int] = mapped_column(primary_key=True) + seed_source: Mapped[str | None] = mapped_column(String(64), index=True) task_id: Mapped[int] = mapped_column( ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False, @@ -132,6 +137,7 @@ class TaskPhoto(Base): __tablename__ = "task_photos" id: Mapped[int] = mapped_column(primary_key=True) + seed_source: Mapped[str | None] = mapped_column(String(64), index=True) task_id: Mapped[int] = mapped_column( ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False, diff --git a/backend/app/schemas/order.py b/backend/app/schemas/order.py index c5cff30..342c3dc 100644 --- a/backend/app/schemas/order.py +++ b/backend/app/schemas/order.py @@ -331,6 +331,7 @@ class OrderDailyReceivableRead(BaseModel): class OrderSummary(BaseModel): id: int + write_revision: str = Field(pattern=r"^[0-9a-f]{64}$") source_customer_id: int | None service_contact: OrderServiceContact cat_snapshot: list[OrderCatSnapshot] diff --git a/backend/app/services/dashboard.py b/backend/app/services/dashboard.py index 46c3224..9034cf4 100644 --- a/backend/app/services/dashboard.py +++ b/backend/app/services/dashboard.py @@ -32,6 +32,7 @@ execution_revision, load_execution_task, require_execution_revision, + reserve_execution_revision, ) @@ -334,6 +335,7 @@ def mark_task_photos_sent( if task.photos_sent_at is not None: raise HTTPException(status_code=409, detail="任务照片已经标记发送") + reserve_execution_revision(session, task, expected_revision) task.photos_sent_at = datetime.now(timezone.utc) session.commit() session.expire_all() diff --git a/backend/app/services/demo_data.py b/backend/app/services/demo_data.py index efd3ad8..47fa822 100644 --- a/backend/app/services/demo_data.py +++ b/backend/app/services/demo_data.py @@ -1,11 +1,12 @@ -from sqlalchemy import delete, func, select +from fastapi import HTTPException +from sqlalchemy import delete, or_, select from sqlalchemy.orm import Session from app.models.customer import Cat, Customer -from app.models.order import Order +from app.models.order import Order, OrderCat from app.models.payment import Payment from app.models.system import SystemFlag -from app.models.task import Task +from app.models.task import Task, TaskItem, TaskPhoto from app.schemas.settings import DemoDataCounts @@ -14,23 +15,62 @@ def demo_data_counts(session: Session) -> DemoDataCounts: - customer_id = session.scalar( - select(Customer.id).where(Customer.system_key == DEMO_SYSTEM_KEY) + customer = session.scalar( + select(Customer).where( + Customer.system_key == DEMO_SYSTEM_KEY, + Customer.seed_source == DEMO_SYSTEM_KEY, + ) ) - if customer_id is None: + if customer is None: return DemoDataCounts(customers=0, cats=0, orders=0, tasks=0, payments=0) - order_ids = select(Order.id).where(Order.customer_id == customer_id) + order_ids = list( + session.scalars( + select(Order.id).where( + Order.customer_id == customer.id, + Order.seed_source == DEMO_SYSTEM_KEY, + ) + ) + ) return DemoDataCounts( customers=1, - cats=int( - session.scalar(select(func.count(Cat.id)).where(Cat.customer_id == customer_id)) - or 0 + cats=len( + list( + session.scalars( + select(Cat.id).where( + Cat.customer_id == customer.id, + Cat.seed_source == DEMO_SYSTEM_KEY, + ) + ) + ) + ), + orders=len(order_ids), + tasks=( + len( + list( + session.scalars( + select(Task.id).where( + Task.order_id.in_(order_ids), + Task.seed_source == DEMO_SYSTEM_KEY, + ) + ) + ) + ) + if order_ids + else 0 ), - orders=int(session.scalar(select(func.count()).select_from(order_ids.subquery())) or 0), - tasks=int(session.scalar(select(func.count(Task.id)).where(Task.order_id.in_(order_ids))) or 0), - payments=int( - session.scalar(select(func.count(Payment.id)).where(Payment.order_id.in_(order_ids))) - or 0 + payments=( + len( + list( + session.scalars( + select(Payment.id).where( + Payment.order_id.in_(order_ids), + Payment.seed_source == DEMO_SYSTEM_KEY, + ) + ) + ) + ) + if order_ids + else 0 ), ) @@ -40,15 +80,97 @@ def demo_data_was_cleared(session: Session) -> bool: def clear_demo_data(session: Session) -> DemoDataCounts: - counts = demo_data_counts(session) - customer_id = session.scalar( - select(Customer.id).where(Customer.system_key == DEMO_SYSTEM_KEY) + customer = session.scalar( + select(Customer).where(Customer.system_key == DEMO_SYSTEM_KEY) ) - if customer_id is not None: - order_ids = select(Order.id).where(Order.customer_id == customer_id) - session.execute(delete(Payment).where(Payment.order_id.in_(order_ids))) - session.execute(delete(Order).where(Order.customer_id == customer_id)) - session.execute(delete(Customer).where(Customer.id == customer_id)) + if customer is None: + counts = DemoDataCounts(customers=0, cats=0, orders=0, tasks=0, payments=0) + else: + if customer.seed_source != DEMO_SYSTEM_KEY: + raise HTTPException(status_code=409, detail="演示数据来源无法确认，未执行清理") + + cats = list(session.scalars(select(Cat).where(Cat.customer_id == customer.id))) + orders = list( + session.scalars(select(Order).where(Order.customer_id == customer.id)) + ) + order_ids = [order.id for order in orders] + tasks = ( + list(session.scalars(select(Task).where(Task.order_id.in_(order_ids)))) + if order_ids + else [] + ) + task_ids = [task.id for task in tasks] + task_items = ( + list( + session.scalars( + select(TaskItem).where(TaskItem.task_id.in_(task_ids)) + ) + ) + if task_ids + else [] + ) + task_photos = ( + list( + session.scalars( + select(TaskPhoto).where(TaskPhoto.task_id.in_(task_ids)) + ) + ) + if task_ids + else [] + ) + payments = list( + session.scalars( + select(Payment).where( + or_( + Payment.customer_id == customer.id, + Payment.order_id.in_(order_ids) if order_ids else False, + ) + ) + ) + ) + linked_cat_ids = ( + set( + session.scalars( + select(OrderCat.cat_id).where(OrderCat.order_id.in_(order_ids)) + ) + ) + if order_ids + else set() + ) + marked_cat_ids = {cat.id for cat in cats if cat.seed_source == DEMO_SYSTEM_KEY} + mixed = any( + ( + any(cat.seed_source != DEMO_SYSTEM_KEY for cat in cats), + any(order.seed_source != DEMO_SYSTEM_KEY for order in orders), + any(task.seed_source != DEMO_SYSTEM_KEY for task in tasks), + any(item.seed_source != DEMO_SYSTEM_KEY for item in task_items), + any(photo.seed_source != DEMO_SYSTEM_KEY for photo in task_photos), + any(payment.seed_source != DEMO_SYSTEM_KEY for payment in payments), + not linked_cat_ids.issubset(marked_cat_ids), + ) + ) + if mixed: + raise HTTPException( + status_code=409, + detail="演示客户下存在未标记数据，为避免误删已停止清理", + ) + + counts = DemoDataCounts( + customers=1, + cats=len(cats), + orders=len(orders), + tasks=len(tasks), + payments=len(payments), + ) + payment_ids = [payment.id for payment in payments] + if payment_ids: + session.execute(delete(Payment).where(Payment.id.in_(payment_ids))) + if order_ids: + session.execute(delete(Order).where(Order.id.in_(order_ids))) + cat_ids = [cat.id for cat in cats] + if cat_ids: + session.execute(delete(Cat).where(Cat.id.in_(cat_ids))) + session.execute(delete(Customer).where(Customer.id == customer.id)) flag = session.get(SystemFlag, DEMO_CLEARED_FLAG) if flag is None: session.add(SystemFlag(key=DEMO_CLEARED_FLAG, payload=counts.model_dump())) diff --git a/backend/app/services/manual_locations.py b/backend/app/services/manual_locations.py index 9b04ecc..53f7b50 100644 --- a/backend/app/services/manual_locations.py +++ b/backend/app/services/manual_locations.py @@ -23,6 +23,10 @@ task_has_execution_history, ) from app.services.plans import day_plan_revision, load_day_tasks, require_current_revision +from app.services.order_revisions import ( + bump_task_revision, + reserve_internal_order_revision, +) def _order_options() -> tuple: @@ -75,6 +79,7 @@ def _eligible_order(order: Order, address: str) -> bool: def _apply_order_location( + session: Session, order: Order, *, latitude: Decimal, @@ -84,6 +89,7 @@ def _apply_order_location( adcode: str | None, level: str, ) -> int: + reserve_internal_order_revision(session, order) order.route_latitude = latitude order.route_longitude = longitude order.route_geocode_status = status @@ -97,6 +103,7 @@ def _apply_order_location( task.planned_lat = latitude task.planned_lng = longitude task.estimated_arrival = None + bump_task_revision(task) affected_tasks += 1 return affected_tasks @@ -205,6 +212,7 @@ def update_customer_location( if not _eligible_order(order, address): continue affected_tasks += _apply_order_location( + session, order, latitude=lat, longitude=lng, @@ -257,6 +265,7 @@ def update_order_location( status = "resolved" if automatic_result is not None else "manual" fingerprint = geocode_fingerprint(provider_name, address) affected_tasks = _apply_order_location( + session, order, latitude=lat, longitude=lng, diff --git a/backend/app/services/order_locations.py b/backend/app/services/order_locations.py index c885d46..80b0f02 100644 --- a/backend/app/services/order_locations.py +++ b/backend/app/services/order_locations.py @@ -16,6 +16,29 @@ order_geocode_address, task_has_execution_history, ) +from app.services.order_revisions import ( + bump_task_revision, + reserve_internal_order_revision, +) + + +def _set_unexecuted_task_location( + task: Task, + latitude: Decimal | None, + longitude: Decimal | None, +) -> None: + if task_has_execution_history(task): + return + if ( + task.planned_lat == latitude + and task.planned_lng == longitude + and task.estimated_arrival is None + ): + return + task.planned_lat = latitude + task.planned_lng = longitude + task.estimated_arrival = None + bump_task_revision(task) def clear_order_location(order: Order) -> None: @@ -26,9 +49,7 @@ def clear_order_location(order: Order) -> None: order.route_geocode_adcode = None order.route_geocode_level = None for task in order.tasks: - if not task_has_execution_history(task): - task.planned_lat = None - task.planned_lng = None + _set_unexecuted_task_location(task, None, None) def trusted_order_point(order: Order, services: MapServices) -> GeoPoint | None: @@ -77,12 +98,9 @@ def _mark_geocode_failure( order: Order, *, status: str, - result: GeocodeResult | None = None, ) -> None: + clear_order_location(order) order.route_geocode_status = status - if result is not None: - order.route_geocode_adcode = result.adcode - order.route_geocode_level = result.level def _apply_geocode_result( @@ -105,9 +123,7 @@ def _apply_geocode_result( order.route_geocode_adcode = result.adcode order.route_geocode_level = result.level for task in order.tasks: - if not task_has_execution_history(task): - task.planned_lat = latitude - task.planned_lng = longitude + _set_unexecuted_task_location(task, latitude, longitude) customer_address = ( customer_geocode_address(order.customer) if order.customer is not None else None @@ -142,6 +158,7 @@ def geocode_order( ) if order is None: return "missing_order" + reserve_internal_order_revision(session, order) address = order_geocode_address(order) if not address: clear_order_location(order) @@ -164,9 +181,11 @@ def geocode_order( order.route_geocode_adcode = None order.route_geocode_level = "manual_pin" for task in order.tasks: - if not task_has_execution_history(task): - task.planned_lat = customer.latitude - task.planned_lng = customer.longitude + _set_unexecuted_task_location( + task, + Decimal(str(customer.latitude)), + Decimal(str(customer.longitude)), + ) session.commit() return "manual" @@ -184,7 +203,7 @@ def geocode_order( return "failed" if not geocode_result_matches_address(address, result): - _mark_geocode_failure(order, status="geocode_mismatch", result=result) + _mark_geocode_failure(order, status="geocode_mismatch") session.commit() return "geocode_mismatch" diff --git a/backend/app/services/order_revisions.py b/backend/app/services/order_revisions.py new file mode 100644 index 0000000..ef64fd6 --- /dev/null +++ b/backend/app/services/order_revisions.py @@ -0,0 +1,113 @@ +import hashlib +import json +from typing import Any + +from fastapi import HTTPException +from sqlalchemy import update +from sqlalchemy.orm import Session +from sqlalchemy.orm.attributes import set_committed_value + +from app.models.order import Order +from app.models.task import Task + + +CONFLICT_DETAIL = "订单已在其他页面更新，请刷新后重试" + + +def order_write_revision(order: Order) -> str: + value = f"{order.id}:{order.write_revision_number}" + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def order_payload_hash(payload: Any) -> str: + body = { + "schema": payload.__class__.__name__, + "payload": payload.model_dump(mode="json"), + } + serialized = json.dumps( + body, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() + + +def require_order_revision(order: Order, expected_revision: str) -> None: + if order_write_revision(order) != expected_revision: + raise HTTPException(status_code=409, detail=CONFLICT_DETAIL) + + +def reserve_order_revision( + session: Session, + order: Order, + expected_revision: str, +) -> None: + """Atomically reserve the next order write version inside the current transaction.""" + + require_order_revision(order, expected_revision) + previous = order.write_revision_number + result = session.execute( + update(Order) + .where( + Order.id == order.id, + Order.write_revision_number == previous, + ) + .values(write_revision_number=previous + 1) + .execution_options(synchronize_session=False) + ) + if result.rowcount != 1: + session.rollback() + raise HTTPException(status_code=409, detail=CONFLICT_DETAIL) + set_committed_value(order, "write_revision_number", previous + 1) + + +def reserve_internal_order_revision(session: Session, order: Order) -> None: + previous = order.write_revision_number + result = session.execute( + update(Order) + .where( + Order.id == order.id, + Order.write_revision_number == previous, + ) + .values(write_revision_number=previous + 1) + .execution_options(synchronize_session=False) + ) + if result.rowcount != 1: + session.rollback() + raise HTTPException(status_code=409, detail=CONFLICT_DETAIL) + set_committed_value(order, "write_revision_number", previous + 1) + + +def bump_task_revision(task: Task) -> None: + task.execution_revision_number += 1 + + +def reserve_internal_task_revision(session: Session, task: Task) -> None: + """Conditionally bump task and order versions after an aggregate token check.""" + + task_revision = task.execution_revision_number + order_revision = task.order.write_revision_number + task_result = session.execute( + update(Task) + .where( + Task.id == task.id, + Task.execution_revision_number == task_revision, + ) + .values(execution_revision_number=task_revision + 1) + .execution_options(synchronize_session=False) + ) + order_result = session.execute( + update(Order) + .where( + Order.id == task.order_id, + Order.write_revision_number == order_revision, + ) + .values(write_revision_number=order_revision + 1) + .execution_options(synchronize_session=False) + ) + if task_result.rowcount != 1 or order_result.rowcount != 1: + session.rollback() + raise HTTPException(status_code=409, detail="任务计划已变化，请刷新后重试") + set_committed_value(task, "execution_revision_number", task_revision + 1) + set_committed_value(task.order, "write_revision_number", order_revision + 1) diff --git a/backend/app/services/orders.py b/backend/app/services/orders.py index 1134a75..0c6241b 100644 --- a/backend/app/services/orders.py +++ b/backend/app/services/orders.py @@ -23,6 +23,7 @@ OrderServiceContact, OrderWrite, ) +from app.services.order_revisions import bump_task_revision MONEY = Decimal("0.01") @@ -686,6 +687,37 @@ def order_has_execution_history(order: Order) -> bool: return any(task_has_execution_history(task) for task in order.tasks) +def require_order_status_transition(order: Order, new_status: OrderStatus) -> None: + if new_status is order.order_status: + return + if new_status in {OrderStatus.IN_PROGRESS, OrderStatus.COMPLETED}: + raise HTTPException( + status_code=409, + detail="进行中和已完成状态只能由任务执行结果生成", + ) + if order.order_status in {OrderStatus.IN_PROGRESS, OrderStatus.COMPLETED}: + raise HTTPException( + status_code=409, + detail="执行中或已完成订单不能人工变更状态", + ) + if order_has_execution_history(order): + raise HTTPException(status_code=409, detail="订单已有执行记录，不能人工变更状态") + + allowed = { + OrderStatus.PENDING_CONFIRMATION: { + OrderStatus.CONFIRMED, + OrderStatus.CANCELLED, + }, + OrderStatus.CONFIRMED: { + OrderStatus.PENDING_CONFIRMATION, + OrderStatus.CANCELLED, + }, + OrderStatus.CANCELLED: {OrderStatus.CONFIRMED}, + } + if new_status not in allowed.get(order.order_status, set()): + raise HTTPException(status_code=409, detail="不允许的订单状态变更") + + def require_tasks_are_rebuildable(order: Order) -> None: if order_has_execution_history(order): raise HTTPException( @@ -705,5 +737,6 @@ def synchronize_task_statuses(order: Order, new_status: OrderStatus) -> None: target_status = initial_task_status(new_status) for task in order.tasks: - if task.status in MUTABLE_TASK_STATUSES: + if task.status in MUTABLE_TASK_STATUSES and task.status is not target_status: task.status = target_status + bump_task_revision(task) diff --git a/backend/app/services/payments.py b/backend/app/services/payments.py index a277eac..c3d00f2 100644 --- a/backend/app/services/payments.py +++ b/backend/app/services/payments.py @@ -348,6 +348,7 @@ def register_payment( update(Order) .where( Order.id == order.id, + Order.write_revision_number == order.write_revision_number, Order.total_amount == order.total_amount, Order.paid_amount == order.paid_amount, Order.payment_status == order.payment_status, @@ -357,7 +358,11 @@ def register_payment( Order.adjustment_amount == order.adjustment_amount, Order.adjustment_service_date == order.adjustment_service_date, ) - .values(paid_amount=next_paid, payment_status=next_status) + .values( + paid_amount=next_paid, + payment_status=next_status, + write_revision_number=order.write_revision_number + 1, + ) .execution_options(synchronize_session=False) ) if result.rowcount != 1: @@ -472,6 +477,7 @@ def void_payment( update(Order) .where( Order.id == order.id, + Order.write_revision_number == order.write_revision_number, Order.total_amount == order.total_amount, Order.paid_amount == previous_paid, Order.payment_status == previous_status, @@ -481,7 +487,11 @@ def void_payment( Order.adjustment_amount == order.adjustment_amount, Order.adjustment_service_date == order.adjustment_service_date, ) - .values(paid_amount=next_paid, payment_status=next_status) + .values( + paid_amount=next_paid, + payment_status=next_status, + write_revision_number=order.write_revision_number + 1, + ) .execution_options(synchronize_session=False) ) if order_result.rowcount != 1: diff --git a/backend/app/services/plans.py b/backend/app/services/plans.py index 4289620..5b1c83b 100644 --- a/backend/app/services/plans.py +++ b/backend/app/services/plans.py @@ -12,6 +12,7 @@ from app.models.task import Task from app.schemas.plan import PlanScheduleItem from app.services.orders import task_has_execution_history +from app.services.order_revisions import reserve_internal_task_revision PLANNING_TASK_STATUSES = { @@ -71,6 +72,7 @@ def day_plan_revision(tasks: Sequence[Task]) -> str: "estimated_arrival": _datetime_value(task.estimated_arrival), "sort_order": task.sort_order, "status": task.status.value, + "execution_revision_number": task.execution_revision_number, "customer_map_state": { "id": task.order.customer_id, "geocode_status": task.order.route_geocode_status, @@ -88,6 +90,7 @@ def day_plan_revision(tasks: Sequence[Task]) -> str: else None ), "updated_at": _datetime_value(task.order.updated_at), + "write_revision_number": task.order.write_revision_number, }, "updated_at": _datetime_value(task.updated_at), "started_at": _datetime_value(task.started_at), @@ -153,11 +156,13 @@ def apply_day_schedule( for sort_order, item in enumerate(schedule): task = tasks_by_id[item.task_id] + reserve_internal_task_revision(session, task) task.sort_order = sort_order task.planned_time = item.planned_time task.estimated_arrival = None session.commit() + session.expire_all() return load_day_tasks(session, service_date) @@ -188,6 +193,9 @@ def update_task_planning_status( detail="任务已有执行记录，不能改回计划状态", ) - task.status = task_status + if task.status is not task_status: + reserve_internal_task_revision(session, task) + task.status = task_status session.commit() + session.expire_all() return load_plan_task(session, task_id) diff --git a/backend/app/services/task_execution.py b/backend/app/services/task_execution.py index ed0a871..9d60617 100644 --- a/backend/app/services/task_execution.py +++ b/backend/app/services/task_execution.py @@ -3,9 +3,10 @@ from datetime import datetime, timezone from fastapi import HTTPException, UploadFile -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session, selectinload +from sqlalchemy.orm.attributes import set_committed_value from app.models.enums import OrderStatus, TaskItemType, TaskStatus from app.models.order import Order, OrderCat @@ -69,11 +70,13 @@ def execution_revision(task: Task) -> str: "cat_status": task.cat_status, "exception_notes": task.exception_notes, "updated_at": _datetime_value(task.updated_at), + "revision_number": task.execution_revision_number, }, "order": { "id": task.order.id, "status": task.order.order_status.value, "updated_at": _datetime_value(task.order.updated_at), + "write_revision_number": task.order.write_revision_number, }, "items": [ { @@ -110,6 +113,44 @@ def require_execution_revision(task: Task, expected_revision: str) -> None: ) +def reserve_execution_revision( + session: Session, + task: Task, + expected_revision: str, +) -> None: + """Reserve both task and order versions before changing execution state.""" + + require_execution_revision(task, expected_revision) + task_revision = task.execution_revision_number + order_revision = task.order.write_revision_number + task_result = session.execute( + update(Task) + .where( + Task.id == task.id, + Task.execution_revision_number == task_revision, + ) + .values(execution_revision_number=task_revision + 1) + .execution_options(synchronize_session=False) + ) + order_result = session.execute( + update(Order) + .where( + Order.id == task.order_id, + Order.write_revision_number == order_revision, + ) + .values(write_revision_number=order_revision + 1) + .execution_options(synchronize_session=False) + ) + if task_result.rowcount != 1 or order_result.rowcount != 1: + session.rollback() + raise HTTPException( + status_code=409, + detail="任务或订单已在其他页面更新，请刷新后重试", + ) + set_committed_value(task, "execution_revision_number", task_revision + 1) + set_committed_value(task.order, "write_revision_number", order_revision + 1) + + def task_execution_detail(task: Task) -> TaskExecutionDetail: order = task.order return TaskExecutionDetail( @@ -242,6 +283,7 @@ def start_task( if task.status not in STARTABLE_STATUSES: raise HTTPException(status_code=409, detail="只有已确认或待出发任务可以开始") + reserve_execution_revision(session, task, expected_revision) task.status = TaskStatus.IN_PROGRESS task.started_at = datetime.now(timezone.utc) task.completed_at = None @@ -266,6 +308,7 @@ def update_task_item( raise HTTPException(status_code=409, detail="拍照事项由实际上传图片自动完成") if item.completed == completed: return task_execution_detail(task) + reserve_execution_revision(session, task, expected_revision) item.completed = completed return _commit_and_reload(session, task) @@ -283,6 +326,7 @@ def update_task_text( _require_in_progress(task) if task.notes == notes and task.cat_status == cat_status: return task_execution_detail(task) + reserve_execution_revision(session, task, expected_revision) task.notes = notes task.cat_status = cat_status return _commit_and_reload(session, task) @@ -306,6 +350,7 @@ async def add_task_photo( task = load_execution_task(session, task_id) require_execution_revision(task, expected_revision) _require_in_progress(task) + reserve_execution_revision(session, task, expected_revision) file_url = persist_uploaded_image(image, task.service_date) try: task.photos.append(TaskPhoto(file_url=file_url)) @@ -336,6 +381,7 @@ def complete_task( if not task.photos: raise HTTPException(status_code=409, detail="必做拍照事项需要至少上传一张图片") + reserve_execution_revision(session, task, expected_revision) task.status = TaskStatus.COMPLETED task.completed_at = datetime.now(timezone.utc) _sync_order_progress(task.order) @@ -351,6 +397,7 @@ def mark_task_exception( task = load_execution_task(session, task_id) require_execution_revision(task, expected_revision) _require_in_progress(task) + reserve_execution_revision(session, task, expected_revision) task.exception_notes = exception_notes task.status = TaskStatus.EXCEPTION task.completed_at = datetime.now(timezone.utc) diff --git a/backend/migrations/versions/0014_consistency_guards.py b/backend/migrations/versions/0014_consistency_guards.py new file mode 100644 index 0000000..f0f1179 --- /dev/null +++ b/backend/migrations/versions/0014_consistency_guards.py @@ -0,0 +1,310 @@ +"""Add order, task, and demo-data consistency guards. + +Revision ID: 0014_consistency_guards +Revises: 0013_intake_audit_events +""" + +from collections.abc import Sequence +from decimal import Decimal + +from alembic import op +import sqlalchemy as sa + + +revision: str = "0014_consistency_guards" +down_revision: str | None = "0013_intake_audit_events" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +DEMO_SYSTEM_KEY = "catcare-demo-seed-v1" + + +def _set_sqlite_foreign_keys(*, enabled: bool) -> None: + connection = op.get_bind() + if connection.dialect.name != "sqlite": + return + with op.get_context().autocommit_block(): + op.execute(sa.text(f"PRAGMA foreign_keys={'ON' if enabled else 'OFF'}")) + + +def _backfill_exact_demo_records() -> None: + """Mark only the unique built-in fixture; ambiguous or mixed rows stay unmarked.""" + + connection = op.get_bind() + customers = sa.table( + "customers", + sa.column("id", sa.Integer()), + sa.column("system_key", sa.String()), + sa.column("name", sa.String()), + sa.column("wechat_name", sa.String()), + sa.column("community", sa.String()), + sa.column("address", sa.Text()), + sa.column("seed_source", sa.String()), + ) + cats = sa.table( + "cats", + sa.column("id", sa.Integer()), + sa.column("customer_id", sa.Integer()), + sa.column("name", sa.String()), + sa.column("breed", sa.String()), + sa.column("personality", sa.Text()), + sa.column("seed_source", sa.String()), + ) + orders = sa.table( + "orders", + sa.column("id", sa.Integer()), + sa.column("customer_id", sa.Integer()), + sa.column("contact_name", sa.String()), + sa.column("cat_count", sa.Integer()), + sa.column("total_amount", sa.Numeric()), + sa.column("notes", sa.Text()), + sa.column("cat_snapshot", sa.JSON()), + sa.column("seed_source", sa.String()), + ) + tasks = sa.table( + "tasks", + sa.column("id", sa.Integer()), + sa.column("order_id", sa.Integer()), + sa.column("customer_id", sa.Integer()), + sa.column("notes", sa.Text()), + sa.column("seed_source", sa.String()), + ) + task_items = sa.table( + "task_items", + sa.column("id", sa.Integer()), + sa.column("task_id", sa.Integer()), + sa.column("item_type", sa.String()), + sa.column("required", sa.Boolean()), + sa.column("completed", sa.Boolean()), + sa.column("seed_source", sa.String()), + ) + task_photos = sa.table( + "task_photos", + sa.column("id", sa.Integer()), + sa.column("task_id", sa.Integer()), + sa.column("seed_source", sa.String()), + ) + payments = sa.table( + "payments", + sa.column("id", sa.Integer()), + sa.column("order_id", sa.Integer()), + sa.column("customer_id", sa.Integer()), + sa.column("amount", sa.Numeric()), + sa.column("notes", sa.Text()), + sa.column("seed_source", sa.String()), + ) + + customer_rows = connection.execute( + sa.select(customers.c.id).where( + customers.c.system_key == DEMO_SYSTEM_KEY, + customers.c.name == "演示客户（虚构）", + customers.c.wechat_name == "演示账号（虚构）", + customers.c.community == "虚构演示小区", + customers.c.address == "仅用于开发演示，不对应任何真实地址", + ) + ).all() + if len(customer_rows) != 1: + return + customer_id = customer_rows[0].id + + cat_rows = connection.execute( + sa.select(cats.c.id, cats.c.name).where( + cats.c.customer_id == customer_id, + cats.c.name.in_(["演示猫咪一号", "演示猫咪二号"]), + cats.c.breed == "虚构品种", + cats.c.personality == "开发测试用虚构档案", + ) + ).all() + cat_ids = {row.name: row.id for row in cat_rows} + if len(cat_rows) != 2 or set(cat_ids) != {"演示猫咪一号", "演示猫咪二号"}: + return + + order_rows = connection.execute( + sa.select(orders.c.id).where( + orders.c.customer_id == customer_id, + orders.c.contact_name == "演示客户（虚构）", + orders.c.cat_count == 2, + orders.c.total_amount == Decimal("105.00"), + orders.c.notes == "仅用于开发测试的虚构订单", + ) + ).all() + if len(order_rows) != 1: + return + order_id = order_rows[0].id + + task_rows = connection.execute( + sa.select(tasks.c.id).where( + tasks.c.order_id == order_id, + tasks.c.customer_id == customer_id, + tasks.c.notes == "虚构演示任务", + ) + ).all() + task_ids = [row.id for row in task_rows] + item_rows = connection.execute( + sa.select( + task_items.c.id, + task_items.c.task_id, + task_items.c.item_type, + task_items.c.required, + task_items.c.completed, + ).where(task_items.c.task_id.in_(task_ids)) + ).all() + photo_rows = connection.execute( + sa.select(task_photos.c.id).where(task_photos.c.task_id.in_(task_ids)) + ).all() + payment_rows = connection.execute( + sa.select(payments.c.id).where( + payments.c.order_id == order_id, + payments.c.customer_id == customer_id, + payments.c.amount == Decimal("105.00"), + payments.c.notes == "虚构演示收款", + ) + ).all() + expected_items = {"feed", "water", "litter", "photo"} + exact_items = all( + len([row for row in item_rows if row.task_id == task_id]) == 4 + and { + row.item_type for row in item_rows if row.task_id == task_id + } == expected_items + and all( + row.required and not row.completed + for row in item_rows + if row.task_id == task_id + ) + for task_id in task_ids + ) + if ( + len(task_rows) != 3 + or len(payment_rows) != 1 + or not exact_items + or photo_rows + ): + return + + connection.execute( + customers.update() + .where(customers.c.id == customer_id) + .values(seed_source=DEMO_SYSTEM_KEY) + ) + connection.execute( + cats.update() + .where(cats.c.id.in_(list(cat_ids.values()))) + .values(seed_source=DEMO_SYSTEM_KEY) + ) + connection.execute( + orders.update() + .where(orders.c.id == order_id) + .values( + seed_source=DEMO_SYSTEM_KEY, + cat_snapshot=[ + {"source_cat_id": cat_ids["演示猫咪一号"], "name": "演示猫咪一号"}, + {"source_cat_id": cat_ids["演示猫咪二号"], "name": "演示猫咪二号"}, + ], + ) + ) + connection.execute( + tasks.update() + .where(tasks.c.id.in_([row.id for row in task_rows])) + .values(seed_source=DEMO_SYSTEM_KEY) + ) + connection.execute( + task_items.update() + .where(task_items.c.id.in_([row.id for row in item_rows])) + .values(seed_source=DEMO_SYSTEM_KEY) + ) + connection.execute( + payments.update() + .where(payments.c.id == payment_rows[0].id) + .values(seed_source=DEMO_SYSTEM_KEY) + ) + + +def upgrade() -> None: + _set_sqlite_foreign_keys(enabled=False) + with op.batch_alter_table("customers") as batch_op: + batch_op.add_column(sa.Column("seed_source", sa.String(length=64))) + batch_op.create_index("ix_customers_seed_source", ["seed_source"]) + + with op.batch_alter_table("cats") as batch_op: + batch_op.add_column(sa.Column("seed_source", sa.String(length=64))) + batch_op.create_index("ix_cats_seed_source", ["seed_source"]) + + with op.batch_alter_table("orders") as batch_op: + batch_op.add_column(sa.Column("seed_source", sa.String(length=64))) + batch_op.add_column( + sa.Column( + "write_revision_number", + sa.Integer(), + nullable=False, + server_default=sa.text("0"), + ) + ) + batch_op.add_column(sa.Column("idempotency_key", sa.String(length=128))) + batch_op.add_column(sa.Column("idempotency_payload_hash", sa.String(length=64))) + batch_op.create_index("ix_orders_seed_source", ["seed_source"]) + batch_op.create_index( + "ix_orders_idempotency_key", ["idempotency_key"], unique=True + ) + + with op.batch_alter_table("tasks") as batch_op: + batch_op.add_column(sa.Column("seed_source", sa.String(length=64))) + batch_op.add_column( + sa.Column( + "execution_revision_number", + sa.Integer(), + nullable=False, + server_default=sa.text("0"), + ) + ) + batch_op.create_index("ix_tasks_seed_source", ["seed_source"]) + + with op.batch_alter_table("task_items") as batch_op: + batch_op.add_column(sa.Column("seed_source", sa.String(length=64))) + batch_op.create_index("ix_task_items_seed_source", ["seed_source"]) + + with op.batch_alter_table("task_photos") as batch_op: + batch_op.add_column(sa.Column("seed_source", sa.String(length=64))) + batch_op.create_index("ix_task_photos_seed_source", ["seed_source"]) + + with op.batch_alter_table("payments") as batch_op: + batch_op.add_column(sa.Column("seed_source", sa.String(length=64))) + batch_op.create_index("ix_payments_seed_source", ["seed_source"]) + _set_sqlite_foreign_keys(enabled=True) + _backfill_exact_demo_records() + + +def downgrade() -> None: + _set_sqlite_foreign_keys(enabled=False) + with op.batch_alter_table("payments") as batch_op: + batch_op.drop_index("ix_payments_seed_source") + batch_op.drop_column("seed_source") + + with op.batch_alter_table("tasks") as batch_op: + batch_op.drop_index("ix_tasks_seed_source") + batch_op.drop_column("execution_revision_number") + batch_op.drop_column("seed_source") + + with op.batch_alter_table("task_photos") as batch_op: + batch_op.drop_index("ix_task_photos_seed_source") + batch_op.drop_column("seed_source") + + with op.batch_alter_table("task_items") as batch_op: + batch_op.drop_index("ix_task_items_seed_source") + batch_op.drop_column("seed_source") + + with op.batch_alter_table("orders") as batch_op: + batch_op.drop_index("ix_orders_idempotency_key") + batch_op.drop_index("ix_orders_seed_source") + batch_op.drop_column("idempotency_payload_hash") + batch_op.drop_column("idempotency_key") + batch_op.drop_column("write_revision_number") + batch_op.drop_column("seed_source") + + with op.batch_alter_table("cats") as batch_op: + batch_op.drop_index("ix_cats_seed_source") + batch_op.drop_column("seed_source") + + with op.batch_alter_table("customers") as batch_op: + batch_op.drop_index("ix_customers_seed_source") + batch_op.drop_column("seed_source") + _set_sqlite_foreign_keys(enabled=True) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 74c3f41..a081bb4 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,8 +1,12 @@ import os +import re import sys +from uuid import uuid4 from pathlib import Path import pytest +import fastapi.testclient as fastapi_testclient +from starlette.testclient import TestClient as StarletteTestClient from alembic import command from alembic.config import Config @@ -17,6 +21,52 @@ from app.main import app # noqa: E402 +class ContractTestClient(StarletteTestClient): + """Supply current write contracts for legacy workflow tests. + + Focused contract tests can set ``X-Test-Skip-Order-Guards`` to verify that + production endpoints reject a missing header. + """ + + def request(self, method: str, url, **kwargs): # type: ignore[no-untyped-def] + headers = dict(kwargs.pop("headers", {}) or {}) + skip_guards = headers.pop("X-Test-Skip-Order-Guards", None) is not None + path = str(url).split("?", 1)[0] + normalized_method = method.upper() + if ( + not skip_guards + and normalized_method == "POST" + and path.rstrip("/") == "/api/admin/orders" + and "Idempotency-Key" not in headers + ): + headers["Idempotency-Key"] = str(uuid4()) + + guarded = re.fullmatch( + r"/api/admin/orders/(?P\d+)(?:/status|/geocode)?/?", + path, + ) + guard_methods = {"PATCH", "PUT", "DELETE"} + if path.endswith("/geocode"): + guard_methods.add("POST") + if ( + not skip_guards + and guarded is not None + and normalized_method in guard_methods + and "If-Match" not in headers + ): + detail = super().request( + "GET", + f"/api/admin/orders/{guarded.group('order_id')}", + ) + if detail.status_code == 200: + headers["If-Match"] = detail.json()["write_revision"] + + return super().request(method, url, headers=headers, **kwargs) + + +fastapi_testclient.TestClient = ContractTestClient + + def alembic_config(database_url: str) -> Config: config = Config(str(BACKEND_DIR / "alembic.ini")) config.set_main_option("sqlalchemy.url", database_url) diff --git a/backend/tests/test_database_preflight.py b/backend/tests/test_database_preflight.py index 4dc0f86..ef2bf47 100644 --- a/backend/tests/test_database_preflight.py +++ b/backend/tests/test_database_preflight.py @@ -1,4 +1,5 @@ from pathlib import Path +import sqlite3 from alembic import command from alembic.config import Config @@ -94,3 +95,21 @@ def fail_upgrade(*_args, **_kwargs) -> None: backups = list(backup_dir.glob("*.db")) assert len(backups) == 1 assert backups[0].stat().st_size > 0 + + +def test_preflight_stops_when_sqlite_foreign_key_check_fails(tmp_path: Path) -> None: + database_path = tmp_path / "broken-foreign-key.db" + database_url = f"sqlite:///{database_path.as_posix()}" + command.upgrade(alembic_config(database_url), "head") + with sqlite3.connect(database_path) as connection: + connection.execute("PRAGMA foreign_keys=OFF") + connection.execute( + "INSERT INTO tasks (order_id, service_date, sort_order, status) " + "VALUES (999999, '2039-01-01', 0, 'pending')" + ) + connection.commit() + + with pytest.raises(DatabasePreparationError, match="完整性或外键"): + ensure_database_ready(database_url, backup_dir=tmp_path / "backups") + + assert not (tmp_path / "backups").exists() diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py index 278a22d..0ed9fed 100644 --- a/backend/tests/test_migrations.py +++ b/backend/tests/test_migrations.py @@ -556,3 +556,146 @@ def test_p15_migration_backfills_schedule_cat_count_and_internal_seed_marker( assert restored_notes == "[catcare-demo-seed-v1]" finally: engine.dispose() + + +def test_consistency_guard_migration_adds_versions_markers_and_unique_key( + migrated_database_url: str, +) -> None: + engine = build_engine(migrated_database_url) + try: + database = inspect(engine) + assert "seed_source" in {item["name"] for item in database.get_columns("customers")} + assert "seed_source" in {item["name"] for item in database.get_columns("cats")} + order_columns = {item["name"] for item in database.get_columns("orders")} + assert { + "seed_source", + "write_revision_number", + "idempotency_key", + "idempotency_payload_hash", + }.issubset(order_columns) + task_columns = {item["name"] for item in database.get_columns("tasks")} + assert {"seed_source", "execution_revision_number"}.issubset(task_columns) + assert "seed_source" in { + item["name"] for item in database.get_columns("task_items") + } + assert "seed_source" in { + item["name"] for item in database.get_columns("task_photos") + } + assert "seed_source" in {item["name"] for item in database.get_columns("payments")} + + order_indexes = {item["name"]: item for item in database.get_indexes("orders")} + assert order_indexes["ix_orders_idempotency_key"]["unique"] == 1 + finally: + engine.dispose() + + +def test_consistency_guard_migration_backfills_only_exact_historical_demo_seed( + tmp_path, +) -> None: + database_path = tmp_path / "consistency-demo-backfill.db" + database_url = f"sqlite:///{database_path.as_posix()}" + config = alembic_config(database_url) + command.upgrade(config, "0013_intake_audit_events") + engine = build_engine(database_url) + try: + with engine.begin() as connection: + connection.execute( + text( + "INSERT INTO customers " + "(id, system_key, name, wechat_name, community, address, is_repeat_customer) " + "VALUES (1, 'catcare-demo-seed-v1', '演示客户（虚构）', " + "'演示账号（虚构）', '虚构演示小区', '仅用于开发演示，不对应任何真实地址', 0)" + ) + ) + connection.execute( + text( + "INSERT INTO cats " + "(id, customer_id, name, breed, personality, medication_required, is_active) " + "VALUES (1, 1, '演示猫咪一号', '虚构品种', '开发测试用虚构档案', 0, 1), " + "(2, 1, '演示猫咪二号', '虚构品种', '开发测试用虚构档案', 0, 1)" + ) + ) + connection.execute( + text( + "INSERT INTO orders " + "(id, customer_id, contact_name, contact_wechat_name, contact_community, " + "contact_address, cat_snapshot, start_date, end_date, visits_per_day, " + "cat_count, service_items, pricing_mode, settlement_mode, adjustment_type, " + "adjustment_amount, base_price, extra_cat_fee, stairs_fee, other_fee, " + "total_amount, paid_amount, payment_status, order_status, notes) VALUES " + "(1, 1, '演示客户（虚构）', '演示账号（虚构）', '虚构演示小区', " + "'仅用于开发演示，不对应任何真实地址', :cat_snapshot, " + "'2039-02-01', '2039-02-03', 1, 2, " + "'[\"feed\",\"water\",\"litter\",\"photo\"]', 'legacy_components', " + "'order_total', 'none', 0, 30, 5, 0, 0, 105, 105, 'paid', " + "'confirmed', '仅用于开发测试的虚构订单')" + ), + { + "cat_snapshot": json.dumps( + [ + {"source_cat_id": None, "name": "演示猫咪一号"}, + {"source_cat_id": None, "name": "演示猫咪二号"}, + ], + ensure_ascii=False, + ) + }, + ) + for task_id, day in enumerate(("01", "02", "03"), start=1): + connection.execute( + text( + "INSERT INTO tasks " + "(id, order_id, customer_id, service_date, planned_time, sort_order, status, notes) " + "VALUES (:id, 1, 1, :service_date, '09:30:00', :sort_order, " + "'confirmed', '虚构演示任务')" + ), + { + "id": task_id, + "service_date": f"2039-02-{day}", + "sort_order": task_id - 1, + }, + ) + for offset, item_type in enumerate( + ("feed", "water", "litter", "photo"), start=1 + ): + connection.execute( + text( + "INSERT INTO task_items " + "(id, task_id, item_type, required, completed) " + "VALUES (:id, :task_id, :item_type, 1, 0)" + ), + { + "id": (task_id - 1) * 4 + offset, + "task_id": task_id, + "item_type": item_type, + }, + ) + connection.execute( + text( + "INSERT INTO payments " + "(id, order_id, customer_id, amount, payment_method, payment_status, paid_at, notes) " + "VALUES (1, 1, 1, 105, 'wechat', 'completed', CURRENT_TIMESTAMP, " + "'虚构演示收款')" + ) + ) + + command.upgrade(config, "head") + with engine.connect() as connection: + markers = connection.execute( + text( + "SELECT " + "(SELECT seed_source FROM customers WHERE id = 1), " + "(SELECT COUNT(*) FROM cats WHERE seed_source = 'catcare-demo-seed-v1'), " + "(SELECT seed_source FROM orders WHERE id = 1), " + "(SELECT COUNT(*) FROM tasks WHERE seed_source = 'catcare-demo-seed-v1'), " + "(SELECT COUNT(*) FROM task_items WHERE seed_source = 'catcare-demo-seed-v1'), " + "(SELECT seed_source FROM payments WHERE id = 1)" + ) + ).one() + snapshot = connection.execute( + text("SELECT cat_snapshot FROM orders WHERE id = 1") + ).scalar_one() + assert tuple(markers) == ("catcare-demo-seed-v1", 2, "catcare-demo-seed-v1", 3, 12, "catcare-demo-seed-v1") + parsed = json.loads(snapshot) if isinstance(snapshot, str) else snapshot + assert [item["source_cat_id"] for item in parsed] == [1, 2] + finally: + engine.dispose() diff --git a/backend/tests/test_order_api.py b/backend/tests/test_order_api.py index da80397..d5c05d8 100644 --- a/backend/tests/test_order_api.py +++ b/backend/tests/test_order_api.py @@ -719,3 +719,86 @@ def test_each_execution_history_signal_blocks_order_delete( assert response.status_code == 409 assert "只能取消" in response.json()["detail"] assert client.get(f"/api/admin/orders/{created['id']}").json()["deletable"] is False + + +def test_order_create_requires_and_replays_idempotency_key( + order_api_context: OrderApiContext, +) -> None: + client = order_api_context.client + payload = simple_order_payload(service_dates=["2032-09-01"]) + + missing = client.post( + "/api/admin/orders", + json=payload, + headers={"X-Test-Skip-Order-Guards": "1"}, + ) + assert missing.status_code == 422 + + headers = {"Idempotency-Key": "test-order-idempotency-20320901"} + created = client.post("/api/admin/orders", json=payload, headers=headers) + replayed = client.post("/api/admin/orders", json=payload, headers=headers) + conflict = client.post( + "/api/admin/orders", + json={**payload, "notes": "同键不同内容"}, + headers=headers, + ) + + assert created.status_code == 201 + assert replayed.status_code == 200 + assert replayed.headers["Idempotent-Replayed"] == "true" + assert replayed.json()["id"] == created.json()["id"] + assert conflict.status_code == 409 + with order_api_context.session_factory() as session: + assert session.scalar(select(func.count(Order.id))) == 1 + + +def test_order_if_match_rejects_missing_and_stale_writes( + order_api_context: OrderApiContext, +) -> None: + client = order_api_context.client + created = client.post( + "/api/admin/orders", + json=simple_order_payload(service_dates=["2032-10-01"]), + ).json() + path = f"/api/admin/orders/{created['id']}" + + missing = client.patch( + path, + json={"notes": "缺少版本"}, + headers={"X-Test-Skip-Order-Guards": "1"}, + ) + first = client.patch( + path, + json={"notes": "第一次写入"}, + headers={"If-Match": created["write_revision"]}, + ) + stale = client.patch( + path, + json={"notes": "不应覆盖"}, + headers={"If-Match": created["write_revision"]}, + ) + + assert missing.status_code == 422 + assert first.status_code == 200 + assert first.json()["write_revision"] != created["write_revision"] + assert stale.status_code == 409 + assert client.get(path).json()["notes"] == "第一次写入" + + +def test_execution_states_cannot_be_set_by_order_admin( + order_api_context: OrderApiContext, +) -> None: + client = order_api_context.client + created = client.post( + "/api/admin/orders", + json=simple_order_payload(service_dates=["2032-11-01"]), + ).json() + + response = client.patch( + f"/api/admin/orders/{created['id']}/status", + json={"order_status": "in_progress"}, + headers={"If-Match": created["write_revision"]}, + ) + + assert response.status_code == 409 + assert "任务执行结果" in response.json()["detail"] diff --git a/backend/tests/test_p18_workflow.py b/backend/tests/test_p18_workflow.py index fc86874..51abc11 100644 --- a/backend/tests/test_p18_workflow.py +++ b/backend/tests/test_p18_workflow.py @@ -1,5 +1,6 @@ from collections.abc import Generator from dataclasses import dataclass, field +from decimal import Decimal import pytest from fastapi.testclient import TestClient @@ -11,7 +12,17 @@ from app.main import app from app.maps import GeoPoint, GeocodeResult, MapServices from app.maps.factory import UnavailableMapProvider, get_map_services -from app.models import Cat, Customer, Order, Payment, SystemFlag, Task +from app.models import ( + Cat, + Customer, + Order, + Payment, + SystemFlag, + Task, + TaskItem, + TaskItemType, + TaskStatus, +) from app.services.demo_data import DEMO_CLEARED_FLAG @@ -329,10 +340,68 @@ def test_failed_auto_geocode_does_not_rollback_and_retry_resolves( ] +def test_failed_geocode_clears_stale_order_and_unexecuted_task_coordinates( + p18_context: P18Context, +) -> None: + created = p18_context.client.post( + "/api/admin/orders", + json=direct_order_payload(service_dates=["2038-09-01", "2038-09-02"]), + ).json() + with p18_context.session_factory.begin() as session: + order = session.get(Order, created["id"]) + assert order is not None + order.route_latitude = Decimal("22.5000000") + order.route_longitude = Decimal("114.0000000") + order.route_geocode_status = "resolved" + order.route_geocode_fingerprint = "f" * 64 + order.route_geocode_adcode = "440307" + order.route_geocode_level = "门牌号" + tasks = list( + session.scalars( + select(Task).where(Task.order_id == order.id).order_by(Task.id) + ) + ) + for task in tasks: + task.planned_lat = Decimal("22.5000000") + task.planned_lng = Decimal("114.0000000") + tasks[1].status = TaskStatus.IN_PROGRESS + + p18_context.geocoder.results[:] = [None] + response = p18_context.client.post( + f"/api/admin/orders/{created['id']}/geocode" + ) + + assert response.status_code == 200 + failed = response.json() + assert failed["route_geocode_status"] == "failed" + with p18_context.session_factory() as session: + order = session.get(Order, created["id"]) + assert order is not None + assert order.route_latitude is None + assert order.route_longitude is None + assert order.route_geocode_fingerprint is None + assert order.route_geocode_adcode is None + assert order.route_geocode_level is None + tasks = list( + session.scalars( + select(Task).where(Task.order_id == order.id).order_by(Task.id) + ) + ) + assert tasks[0].planned_lat is None and tasks[0].planned_lng is None + assert tasks[1].planned_lat == Decimal("22.5000000") + assert tasks[1].planned_lng == Decimal("114.0000000") + + def test_demo_cleanup_is_exact_and_seed_tombstone_prevents_recreation( p18_context: P18Context, ) -> None: assert seed_database(p18_context.database_url) is True + with p18_context.session_factory() as session: + seeded_order = session.scalar( + select(Order).where(Order.seed_source == "catcare-demo-seed-v1") + ) + assert seeded_order is not None + assert all(item["source_cat_id"] is not None for item in seeded_order.cat_snapshot) unrelated = p18_context.client.post( "/api/admin/customers", json={"name": "P18 保留客户（虚构）"} ).json() @@ -365,3 +434,40 @@ def test_demo_cleanup_is_exact_and_seed_tombstone_prevents_recreation( ) ) == 0 assert session.scalar(select(func.count(Order.id))) == 0 + + +def test_demo_cleanup_fails_closed_when_unmarked_data_is_mixed_in( + p18_context: P18Context, +) -> None: + assert seed_database(p18_context.database_url) is True + with p18_context.session_factory.begin() as session: + task = session.scalar( + select(Task).where(Task.seed_source == "catcare-demo-seed-v1") + ) + assert task is not None + session.add( + TaskItem( + task_id=task.id, + item_type=TaskItemType.OTHER, + required=False, + ) + ) + + before = {} + with p18_context.session_factory() as session: + for model in (Customer, Cat, Order, Task, Payment): + before[model] = session.scalar(select(func.count(model.id))) + + response = p18_context.client.post( + "/api/admin/settings/demo-data/clear", + json={ + "system_key": "catcare-demo-seed-v1", + "confirmation": "永久清除演示数据", + }, + ) + + assert response.status_code == 409 + with p18_context.session_factory() as session: + for model, count in before.items(): + assert session.scalar(select(func.count(model.id))) == count + assert session.get(SystemFlag, DEMO_CLEARED_FLAG) is None diff --git a/backend/tests/test_payment_api.py b/backend/tests/test_payment_api.py index b8d802d..d57bab2 100644 --- a/backend/tests/test_payment_api.py +++ b/backend/tests/test_payment_api.py @@ -307,6 +307,7 @@ def test_register_partial_and_final_payment_updates_order_and_dashboard( assert refreshed["receivables"] == [] assert len(refreshed["records"]) == 2 order_detail = client.get(f"/api/admin/orders/{order['id']}").json() + assert order_detail["write_revision"] != order["write_revision"] assert order_detail["paid_amount"] == "30.00" assert order_detail["due_amount"] == "0.00" assert order_detail["payment_status"] == "paid" diff --git a/backend/tests/test_task_api.py b/backend/tests/test_task_api.py index 2321804..4e26657 100644 --- a/backend/tests/test_task_api.py +++ b/backend/tests/test_task_api.py @@ -4,6 +4,7 @@ from pathlib import Path import pytest +from fastapi import HTTPException from fastapi.testclient import TestClient from PIL import Image from sqlalchemy import select @@ -14,6 +15,11 @@ from app.main import app from app.models import Task, TaskPhoto, TaskStatus from app.services import task_uploads +from app.services.task_execution import ( + execution_revision, + load_execution_task, + update_task_text, +) @dataclass(frozen=True) @@ -146,6 +152,45 @@ def complete_non_photo_items(client: TestClient, detail: dict) -> dict: return detail +def test_task_database_cas_rejects_a_preloaded_stale_session( + task_api_context: TaskApiContext, +) -> None: + order = create_task_order(task_api_context.client, service_items=["feed"]) + task_id = order["tasks"][0]["id"] + started = start(task_api_context.client, task_id) + + with ( + task_api_context.session_factory() as first_session, + task_api_context.session_factory() as stale_session, + ): + first_task = load_execution_task(first_session, task_id) + stale_task = load_execution_task(stale_session, task_id) + revision = execution_revision(first_task) + assert execution_revision(stale_task) == revision == started["revision"] + + saved = update_task_text( + first_session, + task_id, + revision, + notes="第一会话写入", + cat_status=None, + ) + assert saved.notes == "第一会话写入" + + with pytest.raises(HTTPException) as conflict: + update_task_text( + stale_session, + task_id, + revision, + notes="不应覆盖", + cat_status=None, + ) + assert conflict.value.status_code == 409 + stale_session.rollback() + + assert get_task(task_api_context.client, task_id)["notes"] == "第一会话写入" + + def test_execution_detail_start_text_and_revision_protection( task_api_context: TaskApiContext, ) -> None: diff --git a/docs/tasks/P27_DATA_CONSISTENCY_GUARDS.md b/docs/tasks/P27_DATA_CONSISTENCY_GUARDS.md new file mode 100644 index 0000000..64268f0 --- /dev/null +++ b/docs/tasks/P27_DATA_CONSISTENCY_GUARDS.md @@ -0,0 +1,68 @@ +# P27 数据一致性护栏 + +## 背景 + +第二次工程复检确认，当前仍值得处理的最高优先级风险集中在订单、任务、地理编码和演示数据清理的数据一致性边界。本任务是既定三阶段整改计划的第一阶段，必须保持为单独、可审查、可回滚的 PR。 + +## 目标 + +- 为订单写入增加幂等键与乐观并发控制。 +- 为任务执行增加数据库级条件更新，避免并发请求静默覆盖。 +- 让订单执行状态由任务执行结果驱动，并收紧人工状态迁移。 +- 在地理编码失败时清除失效坐标及未执行任务的计划坐标。 +- 精确标记演示数据，混入非演示依赖时拒绝清理。 +- 让本地迁移入口执行 SQLite 备份和完整性检查，并拒绝盲目升级非 SQLite 数据库。 + +## 允许修改范围 + +- `backend/app/models/`、`backend/app/schemas/`、`backend/app/api/`、`backend/app/services/`、`backend/app/db/` +- `backend/migrations/versions/0014_consistency_guards.py` +- `backend/tests/` 中与本任务相关的测试 +- `frontend/src/features/orders/`、`frontend/src/features/tasks/` 及相关测试 +- `scripts/` 中数据库迁移、初始化、演示数据脚本 +- 本任务文件 + +## 禁止修改范围 + +- 公网 Relay/Gateway、鉴权和部署拓扑 +- 财务口径、收款业务规则以及与本任务无关的界面重构 +- 生产数据库、远程服务器、密钥、环境变量内容 +- 审计报告与 Codemap 产物 + +## 已确定实现要求 + +1. 新迁移同时兼容 SQLite 和 PostgreSQL，为演示客户、猫咪、订单、任务、任务事项、任务照片和收款新增来源标记，并新增订单写版本、任务执行版本、订单幂等键和请求哈希。 +2. 演示数据历史回填只能匹配完整且唯一的内置指纹；无法确认时保持未标记。 +3. 创建订单要求 `Idempotency-Key`；同键同载荷重放原结果，同键不同载荷返回 `409`。 +4. 订单变更要求 `If-Match`，使用条件 `UPDATE` 预占新版本；并发冲突返回 `409`，不得留下部分写入。 +5. 任务执行使用条件 `UPDATE` 同时保护任务与所属订单；订单的 `in_progress`、`completed` 仅由任务状态派生。 +6. 客户或来源变更属于结构性变更；存在执行证据或收款时返回 `409`，允许时必须原子同步订单快照、猫咪关联、联系方式和任务。 +7. 地理编码失败必须清除订单旧坐标、定位指纹和未执行任务的计划坐标；已执行任务的实际证据不得更改。 +8. 清理演示数据前发现任何未标记依赖时返回 `409`，且删除数为零。 +9. 手动迁移统一经过预检：SQLite 先备份，迁移前后执行完整性和外键检查；本地脚本对非 SQLite 数据库拒绝盲目升级。 + +## 验收标准 + +- 并发订单写、并发任务写均只能有一个成功，另一个稳定返回 `409`。 +- 创建订单在响应丢失后可使用同一幂等键安全重试，不产生重复订单。 +- 状态历史、任务执行证据和订单派生状态一致。 +- 地理编码失败不保留可被误用的旧计划坐标。 +- 纯演示数据可清理；混合数据不删除任何记录。 +- 全量后端测试、前端测试、lint、类型检查和构建通过。 +- Git diff 不包含范围外修改、敏感信息、调试代码或临时产物。 + +## 测试命令 + +- `.venv\\Scripts\\python.exe -m pytest backend/tests` +- `npm --prefix frontend test -- --run` +- `npm --prefix frontend run lint` +- `npm --prefix frontend run typecheck` +- `npm --prefix frontend run build` + +## 返回格式 + +- 修改摘要与关键设计决定 +- 新增/调整测试及准确结果 +- 数据迁移和回滚风险 +- Git 分支、提交、远端 SHA、PR 和 CI 状态 +- 明确声明未修改生产环境、未合并本 PR diff --git a/frontend/src/features/orders/OrdersPage.test.tsx b/frontend/src/features/orders/OrdersPage.test.tsx index 5535983..ba37d0c 100644 --- a/frontend/src/features/orders/OrdersPage.test.tsx +++ b/frontend/src/features/orders/OrdersPage.test.tsx @@ -77,6 +77,7 @@ const catDefaults = { const summary: OrderSummary = { id: 1, + write_revision: "a".repeat(64), source_customer_id: 1, service_contact: serviceContact, cat_snapshot: [ @@ -464,6 +465,7 @@ it("edits an order and keeps cancellation as a separate protected action", async expect(apiMocks.updateOrder).toHaveBeenCalledWith( 1, expect.objectContaining({ cat_count: 3 }), + "a".repeat(64), ), ); @@ -477,7 +479,11 @@ it("edits an order and keeps cancellation as a separate protected action", async fireEvent.click(screen.getByRole("button", { name: "取消订单" })); await waitFor(() => - expect(apiMocks.updateOrderStatus).toHaveBeenCalledWith(1, "cancelled"), + expect(apiMocks.updateOrderStatus).toHaveBeenCalledWith( + 1, + "cancelled", + "a".repeat(64), + ), ); }); @@ -505,8 +511,12 @@ it("allows a payment-history order to change only unit price with its revision", fireEvent.click(within(dialog).getByRole("button", { name: "每次价格增加 5 元" })); fireEvent.click(within(dialog).getByRole("button", { name: "保存订单" })); - await waitFor(() => expect(apiMocks.updateOrder).toHaveBeenCalledWith(1, { - unit_price: "35.03", - expected_financial_revision: "c".repeat(64), - })); + await waitFor(() => expect(apiMocks.updateOrder).toHaveBeenCalledWith( + 1, + { + unit_price: "35.03", + expected_financial_revision: "c".repeat(64), + }, + "a".repeat(64), + )); }); diff --git a/frontend/src/features/orders/OrdersPage.tsx b/frontend/src/features/orders/OrdersPage.tsx index a492bb0..b5e0eee 100644 --- a/frontend/src/features/orders/OrdersPage.tsx +++ b/frontend/src/features/orders/OrdersPage.tsx @@ -127,6 +127,12 @@ export function OrdersPage({ initialCreate = false }: OrdersPageProps) { const [listCollapsed, setListCollapsed] = useState(false); const [scheduleRefreshKey, setScheduleRefreshKey] = useState(0); const listRequestId = useRef(0); + const createIdempotencyKey = useRef(crypto.randomUUID()); + + function beginCreate() { + createIdempotencyKey.current = crypto.randomUUID(); + setFormMode("create"); + } const applyOrders = useCallback((items: OrderSummary[]) => { setOrders(items); @@ -193,8 +199,12 @@ export function OrdersPage({ initialCreate = false }: OrdersPageProps) { async function handleSave(payload: OrderCreateInput | OrderPatchInput) { const saved = formMode === "edit" && orderDetail - ? await updateOrder(orderDetail.id, payload as OrderPatchInput) - : await createOrder(payload as OrderCreateInput); + ? await updateOrder( + orderDetail.id, + payload as OrderPatchInput, + orderDetail.write_revision, + ) + : await createOrder(payload as OrderCreateInput, createIdempotencyKey.current); setOrderDetail(saved); setSearchParams((current) => { const next = new URLSearchParams(current); @@ -204,6 +214,7 @@ export function OrdersPage({ initialCreate = false }: OrdersPageProps) { setFormMode(null); await refreshOrders(); setScheduleRefreshKey((current) => current + 1); + createIdempotencyKey.current = crypto.randomUUID(); } function selectOrder(orderId: number) { @@ -241,7 +252,11 @@ export function OrdersPage({ initialCreate = false }: OrdersPageProps) { setStatusSaving(true); setPageError(null); try { - const updated = await updateOrderStatus(orderDetail.id, status); + const updated = await updateOrderStatus( + orderDetail.id, + status, + orderDetail.write_revision, + ); setOrderDetail(updated); await refreshOrders(); setScheduleRefreshKey((current) => current + 1); @@ -258,7 +273,7 @@ export function OrdersPage({ initialCreate = false }: OrdersPageProps) { setDeleting(true); setPageError(null); try { - await deleteOrder(orderDetail.id); + await deleteOrder(orderDetail.id, orderDetail.write_revision); closeOrderDetail(); await refreshOrders(); setScheduleRefreshKey((current) => current + 1); @@ -296,7 +311,10 @@ export function OrdersPage({ initialCreate = false }: OrdersPageProps) { setGeocoding(true); setPageError(null); try { - const updated = await retryOrderGeocode(orderDetail.id); + const updated = await retryOrderGeocode( + orderDetail.id, + orderDetail.write_revision, + ); setOrderDetail(updated); await refreshOrders(); setScheduleRefreshKey((current) => current + 1); @@ -331,7 +349,7 @@ export function OrdersPage({ initialCreate = false }: OrdersPageProps) { - {!(["cancelled", "completed"] as OrderStatus[]).includes(orderDetail.order_status) ? : null} + {(["pending_confirmation", "confirmed"] as OrderStatus[]).includes(orderDetail.order_status) ? : null} {orderDetail.is_demo_data ? : }