"""Pre-launch hardening — asset state-machine columns + performance indexes. Revision ID: 115 Revises: 114 Create Date: 2026-05-04 4th pre-launch 9-pass audit (P0/P1/P2 bundle). Adds: 1. ``asset_items.deleted_at`` + index — replaces hard DELETE on the asset table; preserves audit trail across asset_history / asset_operations cascades. Service-layer change in PR. 2. ``asset_items.borrowed_by`` (FK) + ``asset_items.borrow_started_at`` — borrow flow no longer overwrites ``assigned_to``. Backfill below converts the legacy "borrowed_by hidden in assigned_to" rows to the new shape. 3. Seven missing performance indexes that table-scan at ~500-employee load: - approval_requests (applicant_id, status) - approval_requests (status, created_at) - approval_steps (approval_id) - approval_steps (approver_id, status) - auth_entitlements (user_id, revoked_at) - notification_outbox (created_at) - audit_logs (actor_user_id, created_at) Backfill rationale (Round 2 audit fix-up 2026-05-04): Pre-115, ``borrow_asset`` only accepted assets in status ('idle', 'available') — so the pre-bug "original owner" was always NULL. The prior bug was that ``borrow_asset`` wrote ``assigned_to = borrower``, making /my/inventory show the asset to the borrower as if they OWN it (and offboarding scans treat the borrower as the responsible party). Under the new model: borrowed rows must have ``assigned_to`` = original owner (NULL when borrowed from idle stock) ``borrowed_by`` = current borrower ``status`` = 'borrowed' Backfill therefore: every row with ``status='borrowed' AND borrowed_by IS NULL`` had its borrower stored in ``assigned_to`` (the old shape). Move ``assigned_to`` → ``borrowed_by`` and clear ``assigned_to``. (No AssetOperation join needed — the legacy borrow op was created with from_user_id=NULL, so the original join-based backfill was inert anyway. This simpler form correctly migrates 100% of legacy borrowed rows in one statement.) Idempotent — safe to re-run (the WHERE clause excludes already- migrated rows). """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "115" down_revision = "114" branch_labels = None depends_on = None def _table_exists(bind, table: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1", ), {"t": table}, ).scalar()) def _column_exists(bind, table: str, column: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.columns " "WHERE table_schema = DATABASE() " " AND table_name = :t AND column_name = :c LIMIT 1", ), {"t": table, "c": column}, ).scalar()) def _index_exists(bind, table: str, name: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.statistics " "WHERE table_schema = DATABASE() " " AND table_name = :t AND index_name = :i LIMIT 1", ), {"t": table, "i": name}, ).scalar()) def _fk_exists(bind, table: str, fk_name: str) -> bool: """Round 2 audit fix-up: explicit FK existence probe so the column-add and constraint-add halves of the migration can be guarded INDEPENDENTLY. Without this, an aborted run that landed the column but not the FK could never recover the FK on retry (the column-exists guard would skip the whole block).""" return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND CONSTRAINT_NAME = :n " " AND CONSTRAINT_TYPE = 'FOREIGN KEY' LIMIT 1", ), {"t": table, "n": fk_name}, ).scalar()) def _fk_source_col_type( bind, target_table: str, target_column: str, *, length: int = 36, ): """Mirror the FK target's DATA_TYPE + COLLATION (TiDB strictness). Round 2 audit fix-up 2026-05-04: explicit collation fallback (utf8mb4_unicode_ci, the prod TiDB Cloud default for ``identity_persons.id``). Previously when ``information_schema`` returned a NULL collation echo, this helper emitted the column without a collation clause and TiDB picked the connection default (``utf8mb4_bin``) — mismatching the FK target and rejecting the constraint with error 3780. """ row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME FROM " "information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c", ), {"t": target_table, "c": target_column}, ).fetchone() data_type = (row[0] or "char").lower() if row else "char" collation = (row[1] if row else None) or "utf8mb4_unicode_ci" type_cls = sa.CHAR if data_type == "char" else sa.String return type_cls(length, collation=collation) def _ensure_index(bind, table: str, name: str, columns: list[str]) -> None: if not _table_exists(bind, table): return if _index_exists(bind, table, name): return op.create_index(name, table, columns) def _drop_index(bind, table: str, name: str) -> None: if not _table_exists(bind, table): return if _index_exists(bind, table, name): op.drop_index(name, table_name=table) def upgrade() -> None: bind = op.get_bind() # 1. asset_items.deleted_at (soft-delete column). if _table_exists(bind, "asset_items") and not _column_exists( bind, "asset_items", "deleted_at", ): op.add_column( "asset_items", sa.Column( "deleted_at", sa.DateTime(timezone=True), nullable=True, ), ) _ensure_index(bind, "asset_items", "ix_asset_items_deleted_at", ["deleted_at"]) # 2. asset_items.borrowed_by + borrow_started_at. # Round 2 audit fix-up 2026-05-04: each sub-step is independently # guarded so an aborted-mid-block re-run can recover. Previously # the FK creation lived inside the column-add ``if`` branch; if the # process died after add_column but before create_foreign_key, the # next run skipped FK creation entirely. _FK_NAME_BORROWED_BY = "fk_asset_items_borrowed_by_persons" if _table_exists(bind, "asset_items"): person_id_type = _fk_source_col_type(bind, "identity_persons", "id") if not _column_exists(bind, "asset_items", "borrowed_by"): op.add_column( "asset_items", sa.Column("borrowed_by", person_id_type, nullable=True), ) if not _fk_exists(bind, "asset_items", _FK_NAME_BORROWED_BY): op.create_foreign_key( _FK_NAME_BORROWED_BY, source_table="asset_items", referent_table="identity_persons", local_cols=["borrowed_by"], remote_cols=["id"], # SET NULL keeps the asset row alive when its borrower # is hard-deleted (the codebase soft-deletes only, but # GDPR / data-retention may eventually hard-delete and # we want the asset to survive that, surfacing in the # admin "stuck borrow" report rather than vanishing). ondelete="SET NULL", ) if not _column_exists(bind, "asset_items", "borrow_started_at"): op.add_column( "asset_items", sa.Column( "borrow_started_at", sa.DateTime(timezone=True), nullable=True, ), ) _ensure_index(bind, "asset_items", "ix_asset_items_borrowed_by", ["borrowed_by"]) # Backfill borrowed_by from the legacy ``assigned_to`` storage. # Pre-115 ``borrow_asset`` only accepted idle/available assets, so # there's no original owner to recover — the borrower was stored # in ``assigned_to``. Move it to ``borrowed_by`` and clear # ``assigned_to``. ``WHERE borrowed_by IS NULL`` keeps the UPDATE # idempotent on re-run. if _table_exists(bind, "asset_items"): bind.execute(sa.text( """ UPDATE asset_items SET borrowed_by = assigned_to, borrow_started_at = COALESCE( borrow_started_at, assigned_at, updated_at ), assigned_to = NULL, assigned_at = NULL WHERE status = 'borrowed' AND borrowed_by IS NULL AND assigned_to IS NOT NULL """, )) # 3. Performance indexes — approval_requests. _ensure_index( bind, "approval_requests", "ix_approval_requests_applicant_status", ["applicant_id", "status"], ) _ensure_index( bind, "approval_requests", "ix_approval_requests_status_created", ["status", "created_at"], ) # approval_steps. _ensure_index( bind, "approval_steps", "ix_approval_steps_approval_id", ["approval_id"], ) _ensure_index( bind, "approval_steps", "ix_approval_steps_approver_status", ["approver_id", "status"], ) # auth_entitlements composite for /auth/me/permissions hot path. _ensure_index( bind, "auth_entitlements", "ix_auth_entitlements_user_revoked", ["user_id", "revoked_at"], ) # notification_outbox: list-by-created_at. _ensure_index( bind, "notification_outbox", "ix_notification_outbox_created_at", ["created_at"], ) # audit_logs: "actions by person X" forensic queries. _ensure_index( bind, "audit_logs", "ix_audit_logs_actor_created", ["actor_user_id", "created_at"], ) # 4. Seed the asset.borrow_overdue scheduled-job row so the new # required-expected_return_date borrow flow has a daily reaper. _seed_borrow_overdue_job(bind) # 5. Seed the feishu.approval_pending_reaper job so leaver-as- # approver scenarios don't wedge requests forever. _seed_approval_reaper_job(bind) _BORROW_OVERDUE_JOB = "asset.borrow_overdue" _APPROVAL_REAPER_JOB = "feishu.approval_pending_reaper" def _seed_borrow_overdue_job(bind) -> None: if not _table_exists(bind, "scheduled_jobs"): return exists = bind.execute( sa.text( "SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1", ), {"n": _BORROW_OVERDUE_JOB}, ).scalar() if exists: return bind.execute(sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (UUID(), :n, :dn, :desc, :hk, 'cron', NULL, '30 9 * * *', 'UTC', 1, JSON_OBJECT(), 0, 0, NOW(), NOW()) """, ), { "n": _BORROW_OVERDUE_JOB, "dn": "资产借出超期 提醒", "desc": ( "每日 09:30 UTC 扫描 status='borrowed' 且 expected_return_date " "< today 的资产,给原归属人 + 借用人各发提醒。同日内幂等。" ), "hk": _BORROW_OVERDUE_JOB, }) def _seed_approval_reaper_job(bind) -> None: if not _table_exists(bind, "scheduled_jobs"): return exists = bind.execute( sa.text( "SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1", ), {"n": _APPROVAL_REAPER_JOB}, ).scalar() if exists: return bind.execute(sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (UUID(), :n, :dn, :desc, :hk, 'cron', NULL, '0 4 * * *', 'UTC', 1, JSON_OBJECT( 'expire_after_days', 14, 'hard_expire_after_days', 30 ), 0, 0, NOW(), NOW()) """, ), { "n": _APPROVAL_REAPER_JOB, "dn": "Feishu approval pending 回收", "desc": ( "每日 04:00 UTC 扫描 14d+ 的 pending ApprovalRequest," "重新拉 Feishu 实例状态;终态(已取消/拒绝/超时)即对齐;" "30d+ 仍 PENDING 强制 expire,通知申请人。修复审批人离职" "导致请求永久卡住。" ), "hk": _APPROVAL_REAPER_JOB, }) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute(sa.text( "DELETE FROM scheduled_jobs WHERE name IN (:a, :b)", ), {"a": _BORROW_OVERDUE_JOB, "b": _APPROVAL_REAPER_JOB}) _drop_index(bind, "audit_logs", "ix_audit_logs_actor_created") _drop_index(bind, "notification_outbox", "ix_notification_outbox_created_at") _drop_index(bind, "auth_entitlements", "ix_auth_entitlements_user_revoked") _drop_index(bind, "approval_steps", "ix_approval_steps_approver_status") _drop_index(bind, "approval_steps", "ix_approval_steps_approval_id") _drop_index(bind, "approval_requests", "ix_approval_requests_status_created") _drop_index(bind, "approval_requests", "ix_approval_requests_applicant_status") _drop_index(bind, "asset_items", "ix_asset_items_borrowed_by") if _table_exists(bind, "asset_items") and _column_exists( bind, "asset_items", "borrow_started_at", ): op.drop_column("asset_items", "borrow_started_at") if _table_exists(bind, "asset_items") and _column_exists( bind, "asset_items", "borrowed_by", ): # Round 2 audit fix-up 2026-05-04: drop FKs by querying # information_schema for the actual constraint name(s) # rather than relying on a hardcoded name + try/except. A # future hotfix that recreates the FK with a different # auto-generated name would otherwise leak the constraint # silently and downgrade would fail at drop_column with # "needed in a foreign key constraint". fk_rows = bind.execute(sa.text( "SELECT CONSTRAINT_NAME " "FROM information_schema.KEY_COLUMN_USAGE " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = 'asset_items' " " AND COLUMN_NAME = 'borrowed_by' " " AND REFERENCED_TABLE_NAME = 'identity_persons'", )).fetchall() for (fk_name,) in fk_rows: op.drop_constraint( fk_name, "asset_items", type_="foreignkey", ) op.drop_column("asset_items", "borrowed_by") _drop_index(bind, "asset_items", "ix_asset_items_deleted_at") if _table_exists(bind, "asset_items") and _column_exists( bind, "asset_items", "deleted_at", ): op.drop_column("asset_items", "deleted_at")