"""audit_logs — composite index for per-resource history lookups. Revision ID: 180 Revises: 179 Create Date: 2026-05-18 Why this exists --------------- The new ``GET /api/v1/audit/resources/history`` endpoint resolves "every audit row for this DNS record / domain / server / asset / software grant" via WHERE resource_type IN (...) AND resource_id = :id ORDER BY created_at DESC ``audit_logs.resource_id`` was unindexed (the table previously indexed ``trace_id``, ``actor_user_id``, ``action``, ``created_at`` only), so a single per-resource history view full-scanned ``audit_logs`` — fine in dev, but at prod growth (~hundreds of thousands of rows / month once HRIS sync, cloud syncs, scheduled jobs, and self-service flows are all writing) every detail-page open becomes an O(N) read. Composite ``(resource_type, resource_id, created_at)`` is the optimal shape for that query: the leading columns match the equality filters exactly, and the trailing ``created_at`` gives the ORDER BY a range-scan without a separate sort step. Same column also accelerates the existing ``/audit/logs?resource_type=X`` admin filter, which today benefits only from the ``created_at`` index. Idempotency ----------- ``CREATE INDEX IF NOT EXISTS``-equivalent via ``_index_exists`` check (same pattern as migrations 165 / 178). """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "180" down_revision = "179" branch_labels = None depends_on = None def _index_exists(bind, table: str, index: str) -> bool: return 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": index}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() if not _index_exists(bind, "audit_logs", "ix_audit_logs_resource_lookup"): op.create_index( "ix_audit_logs_resource_lookup", "audit_logs", ["resource_type", "resource_id", "created_at"], ) def downgrade() -> None: bind = op.get_bind() if _index_exists(bind, "audit_logs", "ix_audit_logs_resource_lookup"): op.drop_index("ix_audit_logs_resource_lookup", table_name="audit_logs")