M34: enforce domain integrity in PostgreSQL

PROJECT_STATE.md

@@ -2762,3 +2762,16 @@ evidence yet."
 - Validation: isolated booking API suite **12 passed**; backend Ruff clean; frontend
   TypeScript lint and production build passed.
 - Exact next action: add database invariants/indexes and harden the shared public demo reset.
+
+## M34 — database-enforced domain integrity (2026-08-10)
+
+- Added named PostgreSQL checks for booking windows/status/odometers, vehicle state and
+  non-negative counters, quality rule/severity/lifecycle values, outbox state/attempts and
+  audit actor types. Model metadata and migration `c24f6a9d013e` stay aligned.
+- Added workload indexes for vehicle availability windows, quality work queues, outbox
+  retries and audit filtering/entity traces.
+- Validation: migration upgraded from an empty PostgreSQL database to head; the complete
+  deterministic seed loaded with the expected 2/180/50/254/75/40/33/20 counts; four
+  direct invalid-state writes were rejected by their named constraints; Ruff passed.
+- Exact next action: serialize and rate-limit shared demo reset, add production web guards
+  and cache MCP reachability evidence.

backend/alembic/versions/c24f6a9d013e_domain_constraints_indexes.py

@@ -0,0 +1,68 @@
+"""add domain constraints and operational indexes
+
+Revision ID: c24f6a9d013e
+Revises: b913a72e8c14
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "c24f6a9d013e"
+down_revision = "b913a72e8c14"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    checks = (
+        ("bookings", "ck_bookings_status", "status IN ('reserved','active','returned','cancelled','blocked')"),
+        ("bookings", "ck_bookings_time_window", "ends_at > starts_at"),
+        ("bookings", "ck_bookings_start_odometer", "start_odometer_km IS NULL OR start_odometer_km >= 0"),
+        ("bookings", "ck_bookings_end_odometer", "end_odometer_km IS NULL OR end_odometer_km >= 0"),
+        ("vehicles", "ck_vehicles_operational_status", "operational_status IN ('available','rented','cleaning','maintenance','blocked')"),
+        ("vehicles", "ck_vehicles_model_year", "model_year BETWEEN 1900 AND 2100"),
+        ("vehicles", "ck_vehicles_odometer", "odometer_km >= 0"),
+        ("vehicles", "ck_vehicles_next_service", "next_service_km >= 0"),
+        ("vehicles", "ck_vehicles_version", "version >= 1"),
+        ("data_quality_issues", "ck_data_quality_rule_type", "rule_type IN ('possible_duplicate_customer','missing_required_field','odometer_regression','booking_overlap','vehicle_status_conflict')"),
+        ("data_quality_issues", "ck_data_quality_severity", "severity IN ('low','medium','high')"),
+        ("data_quality_issues", "ck_data_quality_status", "status IN ('open','deferred','resolved','rejected')"),
+        ("outbox_events", "ck_outbox_delivery_status", "delivery_status IN ('pending','delivering','succeeded','failed')"),
+        ("outbox_events", "ck_outbox_attempts", "attempts >= 0"),
+        ("audit_events", "ck_audit_actor_type", "actor_type IN ('user','service','system')"),
+    )
+    for table, name, condition in checks:
+        op.create_check_constraint(name, table, condition)
+
+    op.create_index("ix_bookings_vehicle_status_window", "bookings", ["vehicle_id", "status", "starts_at", "ends_at"])
+    op.create_index("ix_data_quality_work_queue", "data_quality_issues", ["status", "due_at", "severity"])
+    op.create_index("ix_outbox_delivery_next_attempt", "outbox_events", ["delivery_status", "next_attempt_at"])
+    op.create_index("ix_audit_action_occurred", "audit_events", ["action", "occurred_at"])
+    op.create_index("ix_audit_entity", "audit_events", ["entity_type", "entity_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_audit_entity", table_name="audit_events")
+    op.drop_index("ix_audit_action_occurred", table_name="audit_events")
+    op.drop_index("ix_outbox_delivery_next_attempt", table_name="outbox_events")
+    op.drop_index("ix_data_quality_work_queue", table_name="data_quality_issues")
+    op.drop_index("ix_bookings_vehicle_status_window", table_name="bookings")
+    for table, name in (
+        ("audit_events", "ck_audit_actor_type"),
+        ("outbox_events", "ck_outbox_attempts"),
+        ("outbox_events", "ck_outbox_delivery_status"),
+        ("data_quality_issues", "ck_data_quality_status"),
+        ("data_quality_issues", "ck_data_quality_severity"),
+        ("data_quality_issues", "ck_data_quality_rule_type"),
+        ("vehicles", "ck_vehicles_version"),
+        ("vehicles", "ck_vehicles_next_service"),
+        ("vehicles", "ck_vehicles_odometer"),
+        ("vehicles", "ck_vehicles_model_year"),
+        ("vehicles", "ck_vehicles_operational_status"),
+        ("bookings", "ck_bookings_end_odometer"),
+        ("bookings", "ck_bookings_start_odometer"),
+        ("bookings", "ck_bookings_time_window"),
+        ("bookings", "ck_bookings_status"),
+    ):
+        op.drop_constraint(name, table, type_="check")

backend/app/models/audit.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import datetime
 
-from sqlalchemy import DateTime, String
+from sqlalchemy import CheckConstraint, DateTime, Index, String
 from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -13,6 +13,11 @@ ACTOR_TYPES = ("user", "service", "system")
 
 class AuditEvent(UUIDPrimaryKeyMixin, Base):
     __tablename__ = "audit_events"
+    __table_args__ = (
+        CheckConstraint("actor_type IN ('user','service','system')", name="ck_audit_actor_type"),
+        Index("ix_audit_action_occurred", "action", "occurred_at"),
+        Index("ix_audit_entity", "entity_type", "entity_id"),
+    )
 
     actor_type: Mapped[str] = mapped_column(String(20), nullable=False)
     actor_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True))

backend/app/models/booking.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import datetime
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String
+from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, Index, Integer, String
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -13,6 +13,22 @@ BOOKING_STATUSES = ("reserved", "active", "returned", "cancelled", "blocked")
 
 class Booking(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     __tablename__ = "bookings"
+    __table_args__ = (
+        CheckConstraint(
+            "status IN ('reserved','active','returned','cancelled','blocked')",
+            name="ck_bookings_status",
+        ),
+        CheckConstraint("ends_at > starts_at", name="ck_bookings_time_window"),
+        CheckConstraint(
+            "start_odometer_km IS NULL OR start_odometer_km >= 0",
+            name="ck_bookings_start_odometer",
+        ),
+        CheckConstraint(
+            "end_odometer_km IS NULL OR end_odometer_km >= 0",
+            name="ck_bookings_end_odometer",
+        ),
+        Index("ix_bookings_vehicle_status_window", "vehicle_id", "status", "starts_at", "ends_at"),
+    )
 
     public_ref: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
     customer_id: Mapped[uuid.UUID] = mapped_column(

backend/app/models/data_quality.py

@@ -2,7 +2,7 @@ import uuid
 from datetime import datetime
 from typing import TYPE_CHECKING
 
-from sqlalchemy import DateTime, ForeignKey, String
+from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, String
 from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
@@ -25,6 +25,19 @@ ISSUE_STATUSES = ("open", "deferred", "resolved", "rejected")
 
 class DataQualityIssue(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     __tablename__ = "data_quality_issues"
+    __table_args__ = (
+        CheckConstraint(
+            "rule_type IN ('possible_duplicate_customer','missing_required_field',"
+            "'odometer_regression','booking_overlap','vehicle_status_conflict')",
+            name="ck_data_quality_rule_type",
+        ),
+        CheckConstraint("severity IN ('low','medium','high')", name="ck_data_quality_severity"),
+        CheckConstraint(
+            "status IN ('open','deferred','resolved','rejected')",
+            name="ck_data_quality_status",
+        ),
+        Index("ix_data_quality_work_queue", "status", "due_at", "severity"),
+    )
 
     public_ref: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
     rule_type: Mapped[str] = mapped_column(String(40), nullable=False)

backend/app/models/outbox.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import datetime
 
-from sqlalchemy import DateTime, Integer, String, Text
+from sqlalchemy import CheckConstraint, DateTime, Index, Integer, String, Text
 from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -32,6 +32,14 @@ def is_demo_scenario_failure(event: "OutboxEvent") -> bool:
 
 class OutboxEvent(TimestampMixin, Base):
     __tablename__ = "outbox_events"
+    __table_args__ = (
+        CheckConstraint(
+            "delivery_status IN ('pending','delivering','succeeded','failed')",
+            name="ck_outbox_delivery_status",
+        ),
+        CheckConstraint("attempts >= 0", name="ck_outbox_attempts"),
+        Index("ix_outbox_delivery_next_attempt", "delivery_status", "next_attempt_at"),
+    )
 
     event_id: Mapped[uuid.UUID] = mapped_column(
         UUID(as_uuid=True), primary_key=True, default=uuid.uuid4

backend/app/models/vehicle.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Boolean, Integer, String
+from sqlalchemy import Boolean, CheckConstraint, Integer, String
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base
@@ -9,6 +9,16 @@ OPERATIONAL_STATUSES = ("available", "rented", "cleaning", "maintenance", "block
 
 class Vehicle(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     __tablename__ = "vehicles"
+    __table_args__ = (
+        CheckConstraint(
+            "operational_status IN ('available','rented','cleaning','maintenance','blocked')",
+            name="ck_vehicles_operational_status",
+        ),
+        CheckConstraint("model_year BETWEEN 1900 AND 2100", name="ck_vehicles_model_year"),
+        CheckConstraint("odometer_km >= 0", name="ck_vehicles_odometer"),
+        CheckConstraint("next_service_km >= 0", name="ck_vehicles_next_service"),
+        CheckConstraint("version >= 1", name="ck_vehicles_version"),
+    )
 
     public_ref: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
     make: Mapped[str] = mapped_column(String(80), nullable=False)

backend/tests/test_database_constraints.py

@@ -0,0 +1,23 @@
+import pytest
+from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
+
+from app.core.db import SessionLocal
+
+
+@pytest.mark.parametrize(
+    ("statement", "constraint_name"),
+    (
+        ("UPDATE vehicles SET odometer_km = -1 WHERE public_ref = 'MO-001'", "ck_vehicles_odometer"),
+        ("UPDATE bookings SET ends_at = starts_at WHERE public_ref = 'BK-DEMO-RETURN'", "ck_bookings_time_window"),
+        ("UPDATE data_quality_issues SET status = 'invented' WHERE public_ref = 'DQ-DEMO-DUPLICATE'", "ck_data_quality_status"),
+        ("UPDATE outbox_events SET attempts = -1", "ck_outbox_attempts"),
+    ),
+)
+def test_database_rejects_invalid_domain_state(statement: str, constraint_name: str) -> None:
+    with SessionLocal() as db:
+        with pytest.raises(IntegrityError) as caught:
+            db.execute(text(statement))
+            db.commit()
+        db.rollback()
+    assert constraint_name in str(caught.value)