"""Dynamic email-group management: mirror membership query + parsed rule.

Revision ID: 424
Revises: 423
Create Date: 2026-07-22

Why this exists
---------------
HR needs to see and edit the membership rules of Google Cloud Identity
**dynamic groups** (the ones whose members are computed from a CEL query
over synced department/office/country attributes). A department rename in
SuccessFactors silently empties any group whose query still names the old
value — the g-all / EMEA all-hands incident. These columns let Helix
mirror each dynamic group's query, keep a parsed structured form for the
HR builder UI, and flag "dead" values (referenced but matching zero live
users) for the drift notifier.

All columns hang off the existing ``email_groups`` row (one dynamic group
== one email_groups row), so no new table. Idempotent per-column adds.
"""
from __future__ import annotations

import json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "424"
down_revision = "423"
branch_labels = None
depends_on = None

TABLE = "email_groups"
_DRIFT_HANDLER = "google.dynamic_group_drift"


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"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


def _table_exists(bind, name: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n"
        ),
        {"n": name},
    ).scalar() is not None


def _add(bind, column: sa.Column) -> None:
    if not _column_exists(bind, TABLE, column.name):
        op.add_column(TABLE, column)


def upgrade() -> None:
    bind = op.get_bind()
    _add(bind, sa.Column("membership_query", sa.Text(), nullable=True))
    _add(bind, sa.Column("dynamic_rule", sa.JSON(), nullable=True))
    _add(bind, sa.Column("dynamic_dead_values", sa.JSON(), nullable=True))
    _add(
        bind,
        sa.Column(
            "alert_on_missing_values",
            sa.Boolean(),
            nullable=False,
            server_default="0",
        ),
    )
    _add(bind, sa.Column("drift_signature", sa.String(64), nullable=True))

    # Seed the daily drift-detection cron. IT-only + dedup-gated (only
    # notifies when a group's dead/missing value set changes) + no-op on
    # zero drift, so it satisfies the "new push surface must be IT-only"
    # red line without a storm. Seeded ENABLED: it is the root-cause fix
    # for the g-all/EMEA silent-drop incident this feature addresses.
    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text(
                "INSERT INTO scheduled_jobs "
                "(id, name, display_name, description, handler_key, "
                " trigger_type, cron_expr, timezone, enabled, config, "
                " run_count, failure_count, created_at, updated_at) "
                "SELECT :id, :name, :display, :desc, :handler, 'cron', "
                ":cron, 'UTC', 1, :config, 0, 0, NOW(), NOW() "
                "FROM DUAL WHERE NOT EXISTS "
                "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)"
            ),
            {
                "id": str(uuid.uuid4()),
                "name": _DRIFT_HANDLER,
                "display": "Google 动态邮件组 · 规则漂移检测",
                "desc": (
                    "【IT-only】每天检查动态邮件组规则是否引用了已 0 人的部门/"
                    "属性值(死值→成员静默缺失),漂移集合变化时才推送 IT/HR。"
                ),
                "handler": _DRIFT_HANDLER,
                # Daily 02:10 UTC — after the 01:00 dept schema push so
                # value counts reflect the freshest customSchema values.
                "cron": "10 2 * * *",
                "config": json.dumps({}),
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :h"),
            {"h": _DRIFT_HANDLER},
        )
    for col in (
        "drift_signature",
        "alert_on_missing_values",
        "dynamic_dead_values",
        "dynamic_rule",
        "membership_query",
    ):
        if _column_exists(bind, TABLE, col):
            op.drop_column(TABLE, col)