Reconcile permanent client document UDIN and AQMM schema

alembic/versions/20260711_reconcile_permanent_client_document_udin_aqmm.py

@@ -0,0 +1,131 @@
+"""Reconcile permanent client document UDIN, release and AQMM columns.
+
+Revision ID: 20260711_reconcile_perm_doc_udin_aqmm
+Revises: 20260627_phase_5_engagement_closure_checklist
+Create Date: 2026-07-11
+
+This migration closes a schema gap in which the current
+PermanentClientDocument SQLAlchemy model contains the Phase 1 UDIN/final-release
+fields and Phase 4 AQMM evidence fields, while the original migrations added
+those fields only to engagement_documents.
+
+The upgrade is deliberately idempotent:
+- the table is never dropped or recreated;
+- existing rows and document versions are preserved;
+- only absent columns and indexes are added;
+- no existing column, index or constraint is changed.
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "20260711_reconcile_perm_doc_udin_aqmm"
+down_revision = "20260627_phase_5_engagement_closure_checklist"
+branch_labels = None
+depends_on = None
+
+TABLE_NAME = "permanent_client_documents"
+
+COLUMNS = [
+    # Phase 1 - UDIN and final document release controls.
+    sa.Column("udin_required", sa.Boolean(), nullable=False, server_default=sa.false()),
+    sa.Column("udin_status", sa.String(length=30), nullable=False, server_default="not_required"),
+    sa.Column("udin_number", sa.String(length=30), nullable=True),
+    sa.Column("udin_date", sa.Date(), nullable=True),
+    sa.Column("udin_document_date", sa.Date(), nullable=True),
+    sa.Column("udin_document_type", sa.String(length=120), nullable=True),
+    sa.Column("udin_financial_year", sa.String(length=9), nullable=True),
+    sa.Column("udin_amount", sa.Numeric(14, 2), nullable=True),
+    sa.Column("udin_generated_by_user_id", sa.Integer(), nullable=True),
+    sa.Column("udin_verified_at_utc", sa.DateTime(timezone=True), nullable=True),
+    sa.Column("udin_notes", sa.Text(), nullable=True),
+    sa.Column("final_release_status", sa.String(length=30), nullable=False, server_default="draft"),
+    sa.Column("released_by_user_id", sa.Integer(), nullable=True),
+    sa.Column("released_at_utc", sa.DateTime(timezone=True), nullable=True),
+    sa.Column("release_notes", sa.Text(), nullable=True),
+
+    # Phase 4 - AQMM evidence metadata.
+    sa.Column("is_aqmm_evidence", sa.Boolean(), nullable=False, server_default=sa.false()),
+    sa.Column("evidence_type", sa.String(length=80), nullable=True),
+    sa.Column("evidence_description", sa.Text(), nullable=True),
+    sa.Column("evidence_status", sa.String(length=30), nullable=False, server_default="not_required"),
+    sa.Column("evidence_review_note", sa.Text(), nullable=True),
+    sa.Column("evidence_reviewed_by_user_id", sa.Integer(), nullable=True),
+    sa.Column("evidence_reviewed_at_utc", sa.DateTime(timezone=True), nullable=True),
+]
+
+INDEXES = [
+    ("ix_permanent_client_documents_udin_required", ["udin_required"]),
+    ("ix_permanent_client_documents_udin_status", ["udin_status"]),
+    ("ix_permanent_client_documents_udin_number", ["udin_number"]),
+    ("ix_permanent_client_documents_udin_date", ["udin_date"]),
+    ("ix_permanent_client_documents_udin_financial_year", ["udin_financial_year"]),
+    ("ix_permanent_client_documents_udin_generated_by_user_id", ["udin_generated_by_user_id"]),
+    ("ix_permanent_client_documents_final_release_status", ["final_release_status"]),
+    ("ix_permanent_client_documents_released_by_user_id", ["released_by_user_id"]),
+    ("ix_permanent_client_documents_is_aqmm_evidence", ["is_aqmm_evidence"]),
+    ("ix_permanent_client_documents_evidence_type", ["evidence_type"]),
+    ("ix_permanent_client_documents_evidence_status", ["evidence_status"]),
+    ("ix_permanent_client_documents_evidence_reviewed_by_user_id", ["evidence_reviewed_by_user_id"]),
+]
+
+
+def _inspector(bind):
+    return sa.inspect(bind)
+
+
+def _has_table(bind, table_name: str) -> bool:
+    return _inspector(bind).has_table(table_name)
+
+
+def _column_names(bind, table_name: str) -> set[str]:
+    return {column["name"] for column in _inspector(bind).get_columns(table_name)}
+
+
+def _index_names(bind, table_name: str) -> set[str]:
+    return {
+        index["name"]
+        for index in _inspector(bind).get_indexes(table_name)
+        if index.get("name")
+    }
+
+
+def _add_column_if_missing(bind, column: sa.Column) -> None:
+    if column.name in _column_names(bind, TABLE_NAME):
+        return
+    with op.batch_alter_table(TABLE_NAME) as batch:
+        batch.add_column(column)
+
+
+def _create_index_if_missing(bind, index_name: str, columns: list[str]) -> None:
+    if index_name in _index_names(bind, TABLE_NAME):
+        return
+    # The column check makes the migration safe even on an unexpectedly partial
+    # legacy schema.
+    existing_columns = _column_names(bind, TABLE_NAME)
+    if all(column in existing_columns for column in columns):
+        op.create_index(index_name, TABLE_NAME, columns, unique=False)
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if not _has_table(bind, TABLE_NAME):
+        # A fresh installation should obtain this table from the baseline/model
+        # chain. Avoid creating an incomplete replacement table here.
+        return
+
+    for column in COLUMNS:
+        _add_column_if_missing(bind, column.copy())
+
+    for index_name, columns in INDEXES:
+        _create_index_if_missing(bind, index_name, columns)
+
+
+
+def downgrade() -> None:
+    # Intentionally non-destructive. This is a production reconciliation
+    # migration and cannot reliably distinguish columns that pre-existed on a
+    # partially repaired database from columns added by this revision. Keeping
+    # the reconciled columns protects existing UDIN, release and AQMM metadata.
+    pass