"""provisioning_tasks.not_before — defer revoke/grant until a future moment.

Revision ID: 221
Revises: 220
Create Date: 2026-05-29

Why this exists
---------------
Production incident 2026-05-28 (庄清苓 Thalie Zhuang, PR-fixed here):

HR set termination_date = 2026-06-29 (30 days in the future). The HRIS
sync correctly captured ``leaver_date=2026-06-29`` on the
``OffboardingProcess`` row — but ``_apply_software_policy`` /
``_apply_account_policy`` immediately created provisioning_tasks with
``status=pending``, and the ``provisioning.dispatch`` cron (every 30s)
picked them up and revoked her Google Workspace account + Atlassian
group memberships **the same minute**, 30 days too early.

The provisioning dispatcher had no concept of "wait until leaver_date" —
it ran every pending task as fast as it could. This column adds that
concept: tasks may carry an explicit "do not execute before <timestamp>"
gate. Offboarding writes ``not_before = end of leaver_date in person's
local TZ``, onboarding (deferred provisioning) writes
``not_before = hire_date at 00:00 local TZ``, and the dispatcher's
SELECT skips any task whose gate hasn't passed.

The same mechanism is symmetric for onboarding: future-dated hires
also lose their immediate Google account creation / welcome email path
and instead get them at the right moment via ``hire_day_service``.

Schema notes
------------
* ``not_before DATETIME(6) NULL`` — UTC timestamp. NULL means "no gate,
  run as soon as the dispatcher sees you" (preserves all existing
  behaviour for tasks created by approval flows, manual admin actions,
  audit-driven remediation — none of which need a gate).
* Index ``ix_provisioning_tasks_status_not_before`` on (status, not_before)
  for the dispatcher's hot-path scan
  (``WHERE status='pending' AND (not_before IS NULL OR not_before <= NOW())``).
* No FK — ``not_before`` is a free-form timestamp resolved at write time.

Backfill
--------
None. All existing rows leave ``not_before`` NULL (preserves current
behaviour). Only NEW offboarding/onboarding tasks created after this
migration get the gate written.

Idempotency
-----------
Column existence probed via ``information_schema.COLUMNS``; index
existence via ``information_schema.STATISTICS``. Safe to re-run.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "221"
down_revision = "220"
branch_labels = None
depends_on = None


def _column_exists(bind, table: str, column: str) -> bool:
    return 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() is not 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 = :n LIMIT 1"
        ),
        {"t": table, "n": index},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()

    if not _column_exists(bind, "provisioning_tasks", "not_before"):
        op.add_column(
            "provisioning_tasks",
            sa.Column(
                "not_before",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
        )

    if not _index_exists(
        bind, "provisioning_tasks", "ix_provisioning_tasks_status_not_before",
    ):
        op.create_index(
            "ix_provisioning_tasks_status_not_before",
            "provisioning_tasks",
            ["status", "not_before"],
            unique=False,
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _index_exists(
        bind, "provisioning_tasks", "ix_provisioning_tasks_status_not_before",
    ):
        op.drop_index(
            "ix_provisioning_tasks_status_not_before",
            table_name="provisioning_tasks",
        )
    if _column_exists(bind, "provisioning_tasks", "not_before"):
        op.drop_column("provisioning_tasks", "not_before")