"""Identity Sync Center — outbound employee-data sync framework.

This migration introduces the centralized "data sync hub" the platform
exposes to push curated IdentityPerson fields into downstream systems
(Google Workspace custom schemas, Navan, generic SCIM 2.0 endpoints).

Why a dedicated subsystem instead of re-using the existing
``provisioning_tasks`` queue:

* Provisioning is an *event* (grant / revoke a seat). Identity sync is a
  *continuous projection* — the same person is republished on every
  attribute change (dept, title, location, status). Modeling those as
  one-shot tasks would explode the queue and lose the "current state"
  view admins want.
* Provisioning targets are per-software (one row per (person, software)
  in ``software_seats``). Identity sync targets are per-system and the
  set of synced people is rule-driven (filter), not request-driven.
* The two surfaces have different audit semantics: provisioning audits
  human approval + grant decisions; identity sync audits payload
  diffs and authoritative state.

Tables:

* ``identity_sync_targets`` — one row per outbound system instance.
  Holds filter, mapping, schedule, on_offboard policy, status fields.
* ``identity_sync_state`` — per-(target, person) sync state. Carries
  the last-pushed payload hash so unchanged people are skipped (saves
  thousands of API calls per day at 500 employees × N targets).
* ``identity_sync_runs`` — per-target run history (manual / cron /
  event). We don't reuse ``scheduled_job_runs`` because per-person
  pushes (lifecycle hooks, single-person admin trigger) are first-class
  runs that don't go through a ScheduledJob row.

Bootstrapping:

* Seed the ``identity_sync.dispatch`` ScheduledJob (every minute
  master scheduler — actual per-target cron lives on the target row).
* When a primary ``google_workspace`` integration exists in this
  database, seed a ``google-dept`` target that recreates the legacy
  ``google.dept_sync`` job's behavior (org_path → L1/L2/L3/L4 custom
  schema fields). Disable the old ScheduledJob row so the same work
  doesn't run twice.
"""
from __future__ import annotations

import json
import uuid
from datetime import UTC, datetime

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision = "122_identity_sync_center"
down_revision = "121"
branch_labels = None
depends_on = None


def upgrade() -> None:
    # ── identity_sync_targets ────────────────────────────────────────
    op.create_table(
        "identity_sync_targets",
        sa.Column("id", sa.CHAR(36), primary_key=True),
        # Display + addressing
        sa.Column("name", sa.String(128), nullable=False),
        sa.Column(
            "slug",
            sa.String(64),
            nullable=False,
            comment=(
                "Stable handle. Used as part of the per-target lock key "
                "and master-job logs. Snake-or-kebab case."
            ),
        ),
        sa.Column("description", sa.Text, nullable=True),
        # ``scim`` | ``navan`` | ``google_schema`` (kind controls the
        # runner that consumes the target). ``navan`` is a SCIM preset
        # with default mapping + endpoint; we still keep it as a
        # distinct kind so admins see "Navan" in the UI without having
        # to know it's SCIM internally.
        sa.Column("kind", sa.String(32), nullable=False),
        # FK to ``integrations.id`` carrying credentials. Nullable so an
        # admin can stub a target before creating the integration row
        # (the runner refuses to fire until set).
        sa.Column("integration_id", sa.CHAR(36), nullable=True),
        sa.Column(
            "enabled",
            sa.Boolean(),
            nullable=False,
            server_default="0",
        ),
        # ``deactivate`` (default — SCIM ``active=false``) | ``delete``
        # (DELETE on the user object) | ``keep`` (no-op). Per-target
        # because compliance teams sometimes want delete on contractor
        # systems but deactivate on payroll-adjacent systems.
        sa.Column(
            "on_offboard",
            sa.String(16),
            nullable=False,
            server_default="deactivate",
        ),
        # Mapping rules — list of ``{source_template, transform,
        # value_map, default, target_field}`` entries. See
        # ``identity_sync.mapping`` for the evaluator and the docstring
        # there for the full contract. JSON instead of a side table to
        # keep the entire target definition atomic for revert / preview.
        sa.Column("mapping_config", sa.JSON, nullable=False),
        # Filter rules — `{include_offboarded, departments,
        # include_subtree, countries, employment_statuses,
        # job_title_includes, person_ids, mode (and|or)}`. Filter
        # narrows the candidate pool *before* mapping runs.
        sa.Column("filter_config", sa.JSON, nullable=False),
        # Schedule — ``{trigger: cron|manual|event_only,
        # cron_expr: str|None, timezone: str}``. ``trigger=event_only``
        # disables the master-dispatcher fire path and only runs on
        # lifecycle events / manual triggers.
        sa.Column("schedule_config", sa.JSON, nullable=False),
        # Per-kind options (SCIM endpoint / extra headers / Google
        # schema id / etc.). ``identity_sync.runners`` defines the
        # per-kind shape.
        sa.Column("runner_options", sa.JSON, nullable=True),
        # Run state mirrored here for the list view (the runs table is
        # source of truth).
        sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("last_run_status", sa.String(16), nullable=True),
        sa.Column("last_run_summary", sa.JSON, nullable=True),
        sa.Column(
            "next_run_at",
            sa.DateTime(timezone=True),
            nullable=True,
            comment=(
                "Computed from ``schedule_config.cron_expr`` after each "
                "run. Master dispatcher selects rows with "
                "``next_run_at <= now()`` AND ``enabled=1``."
            ),
        ),
        sa.Column("created_by", sa.CHAR(36), nullable=True),
        sa.Column("updated_by", sa.CHAR(36), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.UniqueConstraint("slug", name="uq_identity_sync_targets_slug"),
        sa.Index("ix_identity_sync_targets_enabled", "enabled"),
        sa.Index("ix_identity_sync_targets_next_run_at", "next_run_at"),
        sa.Index("ix_identity_sync_targets_kind", "kind"),
    )

    # ── identity_sync_state ──────────────────────────────────────────
    # Per-(target, person) state. last_payload_hash is the sha256 of
    # the rendered outbound payload; the runner skips a person whose
    # current rendered payload matches the stored hash, which is the
    # core idempotency mechanism.
    op.create_table(
        "identity_sync_state",
        sa.Column("id", sa.CHAR(36), primary_key=True),
        sa.Column("target_id", sa.CHAR(36), nullable=False),
        sa.Column("person_id", sa.CHAR(36), nullable=False),
        sa.Column(
            "external_id",
            sa.String(255),
            nullable=True,
            comment=(
                "Identifier the downstream system assigned (SCIM "
                "``Users.id``, Navan user id, etc.). Used to address "
                "PUT/PATCH/DELETE without re-resolving by email."
            ),
        ),
        sa.Column(
            "last_payload_hash",
            sa.String(64),
            nullable=True,
            comment="sha256 of the last rendered, sorted-key JSON payload.",
        ),
        sa.Column(
            "last_status",
            sa.String(16),
            nullable=False,
            server_default="pending",
            comment=(
                "pending | success | skipped | failed | deactivated | "
                "deleted | excluded (no longer matches filter)."
            ),
        ),
        sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("last_error", sa.Text, nullable=True),
        sa.Column(
            "retry_count",
            sa.Integer,
            nullable=False,
            server_default="0",
        ),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.UniqueConstraint(
            "target_id", "person_id",
            name="uq_identity_sync_state_target_person",
        ),
        sa.Index(
            "ix_identity_sync_state_target_status",
            "target_id", "last_status",
        ),
        sa.Index("ix_identity_sync_state_person", "person_id"),
    )

    # ── identity_sync_runs ───────────────────────────────────────────
    # One row per per-target run (cron sweep, manual sweep, or
    # per-person push).
    op.create_table(
        "identity_sync_runs",
        sa.Column("id", sa.CHAR(36), primary_key=True),
        sa.Column("target_id", sa.CHAR(36), nullable=False),
        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("duration_ms", sa.Integer, nullable=True),
        sa.Column(
            "status",
            sa.String(16),
            nullable=False,
            comment="running | ok | failed | partial | skipped",
        ),
        sa.Column(
            "trigger_source",
            sa.String(96),
            nullable=False,
            comment=(
                "scheduler | manual:<user_id> | manual:mcp:<user_id> | "
                "event:hire|event:offboard|event:role_change | "
                "person:<person_id>. Max length is dominated by "
                "'manual:mcp:<36-char uuid>' = 47 chars; 96 leaves "
                "headroom for future trigger taxonomies."
            ),
        ),
        # Counters mirrored from the runner summary so the list view
        # doesn't have to parse JSON.
        sa.Column("persons_in_scope", sa.Integer, nullable=True),
        sa.Column("created_count", sa.Integer, nullable=True),
        sa.Column("updated_count", sa.Integer, nullable=True),
        sa.Column("skipped_count", sa.Integer, nullable=True),
        sa.Column("deactivated_count", sa.Integer, nullable=True),
        sa.Column("failed_count", sa.Integer, nullable=True),
        sa.Column("error_message", sa.Text, nullable=True),
        sa.Column("result_summary", sa.JSON, nullable=True),
        sa.Index(
            "ix_identity_sync_runs_target_started",
            "target_id", "started_at",
        ),
    )

    # ── Seed the master dispatcher scheduled job ─────────────────────
    bind = op.get_bind()
    now = datetime.now(UTC)
    dispatcher_id = str(uuid.uuid4())
    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 (
                :id, :name, :display_name, :description, :handler_key,
                'interval', 60, NULL, 'UTC',
                1, NULL, 0, 0,
                :created_at, :updated_at
            )
            """
        ),
        {
            "id": dispatcher_id,
            "name": "identity_sync.dispatch",
            "display_name": "数据同步中心 · 主调度",
            "description": (
                "每 60 秒扫描启用中的同步目标,触发 next_run_at 已到的目标。"
                "实际每个目标的 cron 存在 identity_sync_targets.schedule_config 里。"
                "可以禁用,但禁用后所有定时类同步目标都不会触发(手动 / "
                "事件触发不受影响)。"
            ),
            "handler_key": "identity_sync.dispatch",
            "created_at": now,
            "updated_at": now,
        },
    )

    # ── Migrate the legacy google.dept_sync ScheduledJob ─────────────
    # If the old ``google.dept_sync`` ScheduledJob exists AND a primary
    # google_workspace integration is configured, seed a google-dept
    # target that takes over its work, then disable the old job row.
    # If neither exists (fresh install), nothing happens — the admin
    # creates a target manually after configuring the integration.

    google_integ_row = bind.execute(
        sa.text(
            "SELECT id FROM integrations "
            "WHERE kind = 'software_provisioning' "
            "AND provider = 'google_workspace' "
            "AND enabled = 1 "
            "ORDER BY is_primary DESC, created_at LIMIT 1"
        )
    ).fetchone()

    if google_integ_row is not None:
        # Read the legacy schema-mapping override from platform_settings,
        # if any; fall back to defaults that match the old handler.
        ps_row = bind.execute(
            sa.text(
                "SELECT value FROM platform_settings "
                "WHERE `key` = 'google.dept_schema'"
            )
        ).fetchone()

        # ``platform_settings.value`` is JSON column; SQLAlchemy returns
        # it as a Python object (dict) under MySQL/TiDB JSON type.
        legacy_raw = ps_row[0] if ps_row is not None else None
        if isinstance(legacy_raw, str):
            try:
                legacy_raw = json.loads(legacy_raw)
            except (json.JSONDecodeError, TypeError):
                legacy_raw = None
        legacy_settings: dict = legacy_raw if isinstance(legacy_raw, dict) else {}

        schema_id = legacy_settings.get("schema_id") or "Department_Information"
        fields_map = legacy_settings.get("fields") or {
            "l1": "L1_Dept", "l2": "L2_Dept",
            "l3": "L3_Dept", "l4": "L4_Dept",
        }
        skip_offboarded = bool(legacy_settings.get("skip_offboarded", True))
        drop_root_segments = legacy_settings.get("drop_root_segments") or ["PingCAP"]

        # The mapping for google_schema kind uses a special source kind:
        # ``org_path_level`` — extracts the Nth org_path segment after
        # dropping configured root segments. The runner enforces the
        # special syntax; here we declaratively encode the rule.
        mapping_rules = []
        for level_key in ("l1", "l2", "l3", "l4"):
            target_field = fields_map.get(level_key)
            if not target_field:
                continue
            level_idx = {"l1": 0, "l2": 1, "l3": 2, "l4": 3}[level_key]
            mapping_rules.append({
                "source_template": f"{{org_path_level_{level_idx}}}",
                "transform": None,
                "value_map": None,
                "default": None,
                "target_field": target_field,
            })

        target_id = str(uuid.uuid4())
        bind.execute(
            sa.text(
                """
                INSERT INTO identity_sync_targets (
                    id, name, slug, description, kind, integration_id,
                    enabled, on_offboard,
                    mapping_config, filter_config, schedule_config,
                    runner_options,
                    created_at, updated_at
                ) VALUES (
                    :id, :name, :slug, :description, 'google_schema', :integration_id,
                    1, 'keep',
                    :mapping_config, :filter_config, :schedule_config,
                    :runner_options,
                    :created_at, :updated_at
                )
                """
            ),
            {
                "id": target_id,
                "name": "Google Workspace · 部门字段",
                "slug": "google-dept",
                "description": (
                    "把每位员工的部门层级 (L1/L2/L3/L4) 写入 Google "
                    "Workspace 自定义 Schema 字段。继承自旧版 "
                    "google.dept_sync ScheduledJob,从 Identity Sync "
                    "Center 接管。可以在 UI 里改 schema_id、字段名、"
                    "丢弃根节点列表、过滤范围、调度。"
                ),
                "integration_id": str(google_integ_row[0]),
                "mapping_config": json.dumps({
                    "rules": mapping_rules,
                }),
                "filter_config": json.dumps({
                    "include_offboarded": not skip_offboarded,
                    "departments": [],
                    "include_subtree": True,
                    "countries": [],
                    "employment_statuses": (
                        ["active", "on_leave"]
                        if skip_offboarded
                        else ["active", "on_leave", "offboarded"]
                    ),
                    "job_title_includes": [],
                    "person_ids": [],
                    "mode": "and",
                }),
                "schedule_config": json.dumps({
                    "trigger": "cron",
                    # Daily at 02:00 UTC — same cadence the old
                    # ScheduledJob ran at.
                    "cron_expr": "0 2 * * *",
                    "timezone": "UTC",
                }),
                "runner_options": json.dumps({
                    "schema_id": schema_id,
                    "drop_root_segments": drop_root_segments,
                }),
                "created_at": now,
                "updated_at": now,
            },
        )

        # Delete the legacy ScheduledJob row outright (cascades to its
        # ``scheduled_job_runs`` history via the ondelete=CASCADE FK).
        # Earlier drafts only flipped enabled=0, leaving the row in
        # place — but that required keeping a "deprecated stub" handler
        # registered to placate the framework if someone re-enabled
        # it, and a stub that returns fake success is exactly the
        # kind of misleading no-op that masks production bugs. Clean
        # delete + no stub handler = re-enable attempts fail loudly.
        bind.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE name = 'google.dept_sync'")
        )


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

    bind.execute(
        sa.text("DELETE FROM scheduled_jobs WHERE name = 'identity_sync.dispatch'")
    )

    op.drop_table("identity_sync_runs")
    op.drop_table("identity_sync_state")
    op.drop_table("identity_sync_targets")

    # NOTE: downgrade does NOT recreate the legacy
    # ``google.dept_sync`` ScheduledJob row. Recovery path on
    # downgrade is to re-seed it via migration 045 if needed; this
    # downgrade focuses on cleanly removing the new identity_sync_*
    # tables, not full bidirectional state restoration.