M48: harden demo operations and offsite recovery

backend/app/services/data_quality.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import uuid
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
-from difflib import SequenceMatcher
 
 from sqlalchemy import func, select, update
 from sqlalchemy.orm import Session
@@ -15,6 +14,7 @@ from app.models.data_quality import DataQualityIssue
 from app.models.vehicle import Vehicle
 from app.schemas import CurrentUser, ResolveOdometerRegressionRequest
 from app.services.audit import record_audit_event
+from app.services.data_quality_duplicate_scan import scan_duplicate_customers
 from app.services.vehicle_status import (
     RECOMMENDATION_CODE_NO_CONFLICT,
     VehicleStatusRecommendation,
@@ -25,7 +25,6 @@ from app.services.vehicle_status import (
 
 REQUIRED_CUSTOMER_FIELDS = ("first_name", "last_name")
 REQUIRED_VEHICLE_FIELDS = ("registration_number", "make", "model", "location")
-DUPLICATE_THRESHOLD = 70
 DATA_QUALITY_SCAN_LOCK_ID = 6_138_493_717_091_029_491
 
 
@@ -46,10 +45,6 @@ class ScanResult:
         self.created[rule_type] = self.created.get(rule_type, 0) + 1
 
 
-def _normalize(value: str | None) -> str:
-    return (value or "").strip().lower()
-
-
 def _has_open_issue(db: Session, rule_type: str, entity_type: str, entity_id: uuid.UUID) -> bool:
     return (
         db.scalar(
@@ -129,71 +124,6 @@ def _open_issue(
     scan.bump(rule_type)
 
 
-def _scan_duplicate_customers(db: Session, scan: ScanResult) -> None:
-    customers = list(
-        db.scalars(
-            select(Customer).where(
-                Customer.merged_into_customer_id.is_(None),
-                Customer.anonymized_at.is_(None),
-            )
-        ).all()
-    )
-    customers.sort(key=lambda c: c.public_ref)
-    # The threshold cannot be reached without an exact email (60 points) or phone
-    # (50 points). Block on those normalized identifiers first, so similarity scoring
-    # scales with plausible candidates instead of comparing every customer pair.
-    candidate_pairs: set[tuple[int, int]] = set()
-    for attribute in ("email", "phone"):
-        blocks: dict[str, list[int]] = {}
-        for index, customer in enumerate(customers):
-            key = _normalize(getattr(customer, attribute))
-            if key:
-                blocks.setdefault(key, []).append(index)
-        for indices in blocks.values():
-            for offset, left in enumerate(indices):
-                candidate_pairs.update((left, right) for right in indices[offset + 1 :])
-
-    for left, right in sorted(candidate_pairs):
-        a = customers[left]
-        b = customers[right]
-        score = 0
-        signals: list[dict] = []
-        summary_parts: list[str] = []
-        if _normalize(a.email) and _normalize(a.email) == _normalize(b.email):
-            score += 60
-            signals.append({"code": "duplicate.exact_email"})
-            summary_parts.append("exact email")
-        if _normalize(a.phone) and _normalize(a.phone) == _normalize(b.phone):
-            score += 50
-            signals.append({"code": "duplicate.exact_phone"})
-            summary_parts.append("exact phone")
-        if _normalize(a.postal_code) and _normalize(a.postal_code) == _normalize(b.postal_code):
-            score += 10
-            signals.append({"code": "duplicate.same_postal_code"})
-            summary_parts.append("exact postal code")
-        name_a = f"{_normalize(a.first_name)} {_normalize(a.last_name)}"
-        name_b = f"{_normalize(b.first_name)} {_normalize(b.last_name)}"
-        ratio = SequenceMatcher(None, name_a, name_b).ratio()
-        if ratio >= 0.5:
-            score += round(ratio * 30)
-            signals.append({"code": "duplicate.similar_name", "params": {"score": round(ratio, 2)}})
-            summary_parts.append("similar name")
-
-        if score >= DUPLICATE_THRESHOLD:
-            _open_issue(
-                db,
-                scan,
-                rule_type="possible_duplicate_customer",
-                entity_type="customer",
-                entity_id=a.id,
-                severity="high",
-                summary="; ".join(summary_parts) + f" (score {score})",
-                entity_ref=a.public_ref,
-                related_refs=[b.public_ref],
-                signals=signals,
-            )
-
-
 def _scan_missing_required_fields(db: Session, scan: ScanResult) -> None:
     # Anonymised customers have had their contact data removed on purpose; flagging
     # them as "missing required field" would only be resolvable by re-entering PII.
@@ -361,7 +291,7 @@ def run_scan(
     # database boundary so API and n8n triggers cannot both observe an empty condition.
     db.scalar(select(func.pg_advisory_xact_lock(DATA_QUALITY_SCAN_LOCK_ID)))
     scan = ScanResult()
-    _scan_duplicate_customers(db, scan)
+    scan_duplicate_customers(db, scan, _open_issue)
     _scan_missing_required_fields(db, scan)
     _scan_odometer_regressions(db, scan)
     _scan_booking_overlaps(db, scan)

backend/app/services/data_quality_duplicate_scan.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import uuid
+from difflib import SequenceMatcher
+from typing import Protocol, TypeVar
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.customer import Customer
+
+DUPLICATE_THRESHOLD = 70
+
+
+class ScanAccumulator(Protocol):
+    def bump(self, rule_type: str) -> None: ...
+
+
+ScanTypeContra = TypeVar("ScanTypeContra", bound=ScanAccumulator, contravariant=True)
+
+
+class OpenIssue(Protocol[ScanTypeContra]):
+    def __call__(
+        self,
+        db: Session,
+        scan: ScanTypeContra,
+        *,
+        rule_type: str,
+        entity_type: str,
+        entity_id: uuid.UUID,
+        severity: str,
+        summary: str,
+        entity_ref: str,
+        related_refs: list[str],
+        signals: list[dict] | None = None,
+    ) -> None: ...
+
+
+def _normalize(value: str | None) -> str:
+    return (value or "").strip().lower()
+
+
+def scan_duplicate_customers[ScanType: ScanAccumulator](
+    db: Session, scan: ScanType, open_issue: OpenIssue[ScanType]
+) -> None:
+    customers = list(
+        db.scalars(
+            select(Customer).where(
+                Customer.merged_into_customer_id.is_(None),
+                Customer.anonymized_at.is_(None),
+            )
+        ).all()
+    )
+    customers.sort(key=lambda customer: customer.public_ref)
+
+    # The threshold cannot be reached without an exact email (60 points) or phone
+    # (50 points). Block on normalized identifiers so this remains linear for the
+    # overwhelmingly common case and only scores plausible pairs.
+    candidate_pairs: set[tuple[int, int]] = set()
+    for attribute in ("email", "phone"):
+        blocks: dict[str, list[int]] = {}
+        for index, customer in enumerate(customers):
+            key = _normalize(getattr(customer, attribute))
+            if key:
+                blocks.setdefault(key, []).append(index)
+        for indices in blocks.values():
+            for offset, left in enumerate(indices):
+                candidate_pairs.update((left, right) for right in indices[offset + 1 :])
+
+    for left, right in sorted(candidate_pairs):
+        a = customers[left]
+        b = customers[right]
+        score = 0
+        signals: list[dict] = []
+        summary_parts: list[str] = []
+        if _normalize(a.email) and _normalize(a.email) == _normalize(b.email):
+            score += 60
+            signals.append({"code": "duplicate.exact_email"})
+            summary_parts.append("exact email")
+        if _normalize(a.phone) and _normalize(a.phone) == _normalize(b.phone):
+            score += 50
+            signals.append({"code": "duplicate.exact_phone"})
+            summary_parts.append("exact phone")
+        if _normalize(a.postal_code) and _normalize(a.postal_code) == _normalize(b.postal_code):
+            score += 10
+            signals.append({"code": "duplicate.same_postal_code"})
+            summary_parts.append("exact postal code")
+        name_a = f"{_normalize(a.first_name)} {_normalize(a.last_name)}"
+        name_b = f"{_normalize(b.first_name)} {_normalize(b.last_name)}"
+        ratio = SequenceMatcher(None, name_a, name_b).ratio()
+        if ratio >= 0.5:
+            score += round(ratio * 30)
+            signals.append(
+                {"code": "duplicate.similar_name", "params": {"score": round(ratio, 2)}}
+            )
+            summary_parts.append("similar name")
+
+        if score >= DUPLICATE_THRESHOLD:
+            open_issue(
+                db,
+                scan,
+                rule_type="possible_duplicate_customer",
+                entity_type="customer",
+                entity_id=a.id,
+                severity="high",
+                summary="; ".join(summary_parts) + f" (score {score})",
+                entity_ref=a.public_ref,
+                related_refs=[b.public_ref],
+                signals=signals,
+            )

backend/entrypoint.sh

@@ -1,5 +1,7 @@
 #!/bin/sh
 set -e
 
-alembic upgrade head
+if [ "${RUN_MIGRATIONS:-true}" = "true" ]; then
+  alembic upgrade head
+fi
 exec uvicorn app.main:app --host 0.0.0.0 --port 8000