feat(dbmigration): database migration for multi-reviewer schema

backend/api/core/config.py

@@ -139,7 +139,9 @@ def convert_max_file_size(cls, v):
     PDF_LINKAGE_MAX_BYTES: int = int(
         os.getenv('PDF_LINKAGE_MAX_BYTES', str(50 * 1024 * 1024)),
     )
-    PDF_LINKAGE_MAX_RETRIES: int = int(os.getenv('PDF_LINKAGE_MAX_RETRIES', '3'))
+    PDF_LINKAGE_MAX_RETRIES: int = int(
+        os.getenv('PDF_LINKAGE_MAX_RETRIES', '3'),
+    )
     PDF_LINKAGE_MAX_CONCURRENCY: int = int(
         os.getenv('PDF_LINKAGE_MAX_CONCURRENCY', '4'),
     )
@@ -154,6 +156,21 @@ def convert_max_file_size(cls, v):
     )
     DEBUG: bool = os.getenv('DEBUG', 'false').lower() == 'true'
 
+    # Multi-reviewer foundation. Keep disabled until schema and compatibility
+    # verification has passed in the target environment.
+    ENABLE_MULTI_REVIEWER_SCHEMA: bool = os.getenv(
+        'ENABLE_MULTI_REVIEWER_SCHEMA', 'false',
+    ).lower().strip() == 'true'
+    MULTI_REVIEWER_SHADOW_MODE: bool = os.getenv(
+        'MULTI_REVIEWER_SHADOW_MODE', 'false',
+    ).lower().strip() == 'true'
+    # Schema-changing work is performed by the deployment migration command.
+    # This is intentionally false in production; it is available for local
+    # development when an explicit startup bootstrap is desired.
+    AUTO_MIGRATE: bool = os.getenv(
+        'AUTO_MIGRATE', 'true',
+    ).lower().strip() == 'true'
+
     # -------------------------------------------------------------------------
     # Postgres configuration
     # -------------------------------------------------------------------------

backend/api/migrations/__init__.py

@@ -0,0 +1,2 @@
+"""Explicit database migration commands for CAN-SR."""
+from __future__ import annotations

backend/api/migrations/__main__.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+from .cli import main
+
+main()

backend/api/migrations/cli.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import argparse
+
+from api.core.config import settings
+from api.services.review_schema_service import review_schema_service
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description='CAN-SR database migrations')
+    parser.add_argument('command', choices=['migrate', 'verify', 'status'])
+    args = parser.parse_args()
+    if args.command == 'migrate':
+        result = review_schema_service.migrate(settings.VERSION)
+    else:
+        result = review_schema_service.verify_schema()
+    print(result)
+
+
+if __name__ == '__main__':
+    main()

backend/api/screen/router.py

@@ -29,6 +29,11 @@
 from ..services.cit_db_service import cits_dp_service
 from ..services.cit_db_service import snake_case
 from ..services.cit_db_service import snake_case_column
+from ..services.postgres_auth import postgres_server
+from ..services.review_service import reviewer_identity
+from ..services.review_service import ReviewerIdentity
+from ..services.review_service import ReviewRepository
+from ..services.review_service import WorkUnit
 from ..services.screening_eligibility_service import compute_screening_decisions
 from ..services.screening_eligibility_service import screening_eligibility_service
 from ..services.sr_db_service import srdb_service
@@ -1639,6 +1644,33 @@ async def validate_screening_step(
             detail='Citation not found to update',
         )
 
+    # Feature-gated canonical dual-write. The legacy write above remains the
+    # compatibility projection and continues to define the response contract.
+    if settings.ENABLE_MULTI_REVIEWER_SCHEMA:
+        canonical_step = 'extract' if step == 'parameters' else step
+        try:
+            identity = reviewer_identity(current_user)
+            unit = WorkUnit(
+                sr_id=sr_id,
+                stage=canonical_step,
+                source_table_name=table_name,
+                citation_id=citation_id,
+                criteria_revision=int(
+                    (_sr or {}).get('criteria_revision') or 1,
+                ),
+            )
+            repository = ReviewRepository(postgres_server.conn)
+            if checked:
+                repository.validate(unit, identity, {'checked': True})
+            else:
+                repository.remove_current(unit, identity)
+        except Exception as e:
+            logger.exception('Canonical multi-reviewer dual-write failed')
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail=f'Canonical review storage unavailable: {e}',
+            )
+
     return {
         'status': 'success',
         'sr_id': sr_id,

backend/api/services/review_schema_service.py

@@ -0,0 +1,162 @@
+"""Generic migration runner and read-only schema verifier.
+
+Migrations are applied explicitly by deployment (or the CLI), while the API
+startup path only verifies that the database is compatible with the code.
+"""
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Any
+
+MIGRATIONS_PATH = Path(__file__).resolve().parents[2] / 'migrations'
+MIGRATION_TABLE = 'review_schema_migrations'
+# Compatibility aliases retained for callers/tests from the first bootstrap.
+MIGRATION_PATH = MIGRATIONS_PATH / '001_multi_reviewer_schema.sql'
+MIGRATION_VERSION = MIGRATION_PATH.stem
+REQUIRED_TABLES = {
+    'review_schema_migrations', 'review_assignment_policies',
+    'review_assignments', 'review_validations', 'reconciliation_cases',
+    'reconciliation_decisions',
+}
+
+
+def migration_checksum(sql: str | None = None) -> str:
+    text = sql if sql is not None else ''
+    return hashlib.sha256(text.encode('utf-8')).hexdigest()
+
+
+def migration_files() -> list[Path]:
+    return sorted(MIGRATIONS_PATH.glob('[0-9][0-9][0-9]_*.sql'))
+
+
+class ReviewSchemaService:
+    """Apply and verify the additive schema using an injected DB connection."""
+
+    def __init__(self, connection_provider=None):
+        self.connection_provider = connection_provider
+
+    def _provider(self):
+        if self.connection_provider is None:
+            # Import lazily so disabled/shadow-mode tests and application
+            # startup do not require database credentials just to import the
+            # service module.
+            from .postgres_auth import postgres_server
+            self.connection_provider = postgres_server
+        return self.connection_provider
+
+    def _ensure_migration_table(self, cur) -> None:
+        cur.execute(
+            """CREATE TABLE IF NOT EXISTS review_schema_migrations (
+                version TEXT PRIMARY KEY,
+                applied_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                checksum TEXT NOT NULL,
+                app_version TEXT,
+                execution_ms INTEGER
+            )""",
+        )
+        # The first opt-in bootstrap created this table without execution_ms.
+        # Keep upgrades additive so existing installations can adopt the
+        # generic runner without a manual repair step.
+        cur.execute(
+            'ALTER TABLE review_schema_migrations '
+            'ADD COLUMN IF NOT EXISTS execution_ms INTEGER',
+        )
+
+    def migrate(self, app_version: str | None = None) -> dict[str, Any]:
+        """Apply all pending migrations in filename order.
+
+        Each migration is committed independently. A failed migration is
+        rolled back and later migrations are not attempted.
+        """
+        conn = self._provider().conn
+        files = migration_files()
+        if not files:
+            raise RuntimeError(f'No migrations found in {MIGRATIONS_PATH}')
+        applied: list[str] = []
+        for path in files:
+            version = path.stem
+            sql = path.read_text(encoding='utf-8')
+            checksum = hashlib.sha256(sql.encode('utf-8')).hexdigest()
+            cur = conn.cursor()
+            try:
+                cur.execute(
+                    'SELECT pg_advisory_xact_lock(hashtext(%s))',
+                    ('can_sr_schema_migrations',),
+                )
+                self._ensure_migration_table(cur)
+                cur.execute(
+                    f'SELECT checksum FROM {MIGRATION_TABLE} WHERE version = %s',
+                    (version,),
+                )
+                existing = cur.fetchone()
+                if existing:
+                    if existing[0] != checksum:
+                        raise RuntimeError(
+                            f'{version} checksum mismatch; refusing to continue',
+                        )
+                    conn.commit()
+                    continue
+                cur.execute(sql)
+                cur.execute(
+                    f'''INSERT INTO {MIGRATION_TABLE}
+                        (version, checksum, app_version) VALUES (%s, %s, %s)''',
+                    (version, checksum, app_version),
+                )
+                conn.commit()
+                applied.append(version)
+            except Exception:
+                conn.rollback()
+                raise
+            finally:
+                cur.close()
+        return {'applied': applied, **self.verify_schema(connection=conn)}
+
+    # Backward-compatible name for callers of the original opt-in bootstrap.
+    def ensure_schema(self, app_version: str | None = None) -> dict[str, Any]:
+        return self.migrate(app_version)
+
+    def verify_schema(self, connection=None) -> dict[str, Any]:
+        conn = connection or self._provider().conn
+        cur = conn.cursor()
+        try:
+            cur.execute(
+                """SELECT table_name FROM information_schema.tables
+                   WHERE table_schema = 'public'
+                     AND table_name = ANY(%s)""",
+                (sorted(REQUIRED_TABLES),),
+            )
+            tables = {row[0] for row in cur.fetchall()}
+            missing = sorted(REQUIRED_TABLES - tables)
+            if missing:
+                raise RuntimeError(
+                    f'Multi-reviewer schema is incomplete: {missing}',
+                )
+            cur.execute(
+                f'SELECT version FROM {MIGRATION_TABLE} ORDER BY version',
+            )
+            versions = [row[0] for row in cur.fetchall()]
+            expected_versions = [path.stem for path in migration_files()]
+            pending = [
+                version for version in expected_versions if version not in versions
+            ]
+            if pending:
+                raise RuntimeError(f'Database migrations pending: {pending}')
+            # Older installations may retain the first bootstrap marker
+            # (`multi_reviewer_v1`). It remains useful audit history, but must
+            # not be treated as a newer migration than numeric file versions.
+            legacy_versions = [
+                version for version in versions if version not in expected_versions
+            ]
+            return {
+                'version': expected_versions[-1] if expected_versions else None,
+                'versions': versions,
+                'legacy_versions': legacy_versions,
+                'pending': pending,
+                'tables': sorted(tables & REQUIRED_TABLES),
+            }
+        finally:
+            cur.close()
+
+
+review_schema_service = ReviewSchemaService()