fix: 收紧订单与任务数据一致性

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,23 +791,27 @@ 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()
     return _order_detail(_load_order(session, order.id))
 
 
 @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,

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)

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()

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,
         )
     )
 

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,

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,