"""DNS self-service requests + owner recertification. Revision ID: 098 Revises: 097 Create Date: 2026-05-01 ═══════════════════════════════════════════════════════════════════════ What this migration adds ═══════════════════════════════════════════════════════════════════════ (1) New table ``cmdb_dns_record_requests`` ------------------------------------------ Backs the employee-facing self-service flow on ``/my/network-requests``. An employee submits a DNS create / update / delete request which gets routed to a Feishu approval; on approval we push the change into the existing CMDB DNS provider path (which today talks to Tencent DNSPod, AWS Route53, Aliyun, GCP, Azure, UCloud — the same code admins use under ``/admin/cloud/dns``). Status state-machine (kept tight on purpose): pending → approved → executing → executed ↘ failed pending → rejected pending → cancelled (requester withdraws before approval) Once a row reaches ``executed``, ``rejected``, ``cancelled``, or ``failed`` it is terminal; the Feishu callback handler short-circuits duplicates against ``feishu_instance_code`` so a redelivered webhook never double-applies a DNS write. Columns: * identity: ``id``, ``requester_person_id``, ``feishu_instance_code`` (UNIQUE — idempotency key) * what is requested: ``action`` (``create``/``update``/``delete``), ``integration_id``, ``zone_name``, ``name``, ``record_type``, ``value``, ``ttl``, ``line``, ``target_record_id`` (FK on update/delete; NULL on create) * business reason: ``purpose`` (required free-text), ``attachment_url`` * lifecycle: ``status``, ``error_message``, ``executed_record_pk`` (UUID of the DB row created after a successful ``create`` — lets the UI link back to ``/my/cloud/dns`` filtered to this record) * timestamps: ``submitted_at``, ``decided_at``, ``executed_at``, ``created_at``, ``updated_at`` Indexes match the two hot queries: list-by-requester (``my requests`` page) and list-by-zone (admin overview). (2) Column add on ``cmdb_dns_records`` -------------------------------------- * ``last_owner_review_at DATETIME(6) NULL`` — set to NOW() each time the owner clicks "still in use" in the recertification Feishu card. NULL means the record has never been recertified (treated as "due" by the recertification scheduler so existing prod records get nagged on first run). (3) Two scheduled jobs ---------------------- * ``cmdb.dns_owner_recertify`` — daily 09:30 Asia/Shanghai. Finds records where ``last_owner_review_at`` is older than ``platform_settings['cmdb.dns.recertify_period_days']`` (default 90), groups by owner, sends one Feishu card per owner listing all due records with deep-links back to ``/my/cloud/dns``. The card has a "all still in use" button that POSTs to a new ``/cmdb/dns-records/recertify-bulk`` endpoint. * ``cmdb.dns_pre_leaver_handover`` — daily 09:30 Asia/Shanghai. Finds persons whose ``last_working_day`` falls in [today, today+7] and who own at least one DNS record / domain, sends the leaver a Feishu reminder listing the records with the deep-link to submit a delete request now (or rely on the existing T+0 force-transfer fallback). Both jobs are read-only with side-effect notifications — no DB writes to records themselves, so re-running them on the same day produces duplicate notifications but never corrupts state. We accept the duplicate-notification risk in exchange for the simpler implementation (no ``last_notified_stage`` column). (4) Default platform setting ---------------------------- * ``cmdb.dns.recertify_period_days`` = 90 (30-3650 valid range, enforced at the application layer when admin edits via ``/admin/integrations/platform-settings``). ═══════════════════════════════════════════════════════════════════════ Idempotency ═══════════════════════════════════════════════════════════════════════ Every CREATE / ALTER / INSERT here is preceded by an existence guard so re-running ``alembic upgrade head`` is a no-op. The downgrade drops the new table, removes the column, and unregisters the jobs + platform setting (best-effort — the job rows might already be deleted by the admin UI, in which case the DELETE matches zero rows). """ from __future__ import annotations import json import logging import uuid import sqlalchemy as sa from alembic import op logger = logging.getLogger(__name__) revision = "098" down_revision = "097" branch_labels = None depends_on = None _REQUESTS_TABLE = "cmdb_dns_record_requests" _RECORDS_TABLE = "cmdb_dns_records" _NEW_RECORDS_COL = "last_owner_review_at" _JOB_RECERTIFY = "cmdb.dns_owner_recertify" _JOB_LEAVER = "cmdb.dns_pre_leaver_handover" _SETTING_KEY = "cmdb.dns.recertify_period_days" _SETTING_DEFAULT = 90 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 _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 _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 _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "char", length: int = 36, ): """Return a SQLAlchemy column type whose ``DATA_TYPE`` and ``COLLATION_NAME`` match the FK target column on the live DB. Helix's DB has historical drift: ``identity_persons.id`` is CHAR(36) utf8mb4_unicode_ci on prod / local-dev, but CI's fresh MySQL boots it as VARCHAR(36) utf8mb4_0900_ai_ci. MySQL 8 + TiDB both reject FK on mismatch (error 3780). Same pattern used by migrations 091 / 093 / 073 / 088 — query the target's metadata and build a matching child type. """ 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() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] else: data_type = fallback_type.lower() collation = None type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) def _ensure_scheduled_job( bind, *, name: str, display_name: str, description: str, cron_expr: str, handler_key: str, config: dict, timezone: str = "Asia/Shanghai", ) -> 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": name}, ).scalar() if exists: return 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) " "VALUES (:id, :n, :dn, :d, :hk, 'cron', :ce, " " :tz, 1, :cfg, 0, 0, NOW(), NOW())" ), { "id": str(uuid.uuid4()), "n": name, "dn": display_name, "d": description, "hk": handler_key, "ce": cron_expr, "tz": timezone, "cfg": json.dumps(config), }, ) def upgrade() -> None: bind = op.get_bind() # ─── (1) cmdb_dns_record_requests ────────────────────────────── # FK target columns can have any of (CHAR | VARCHAR) × any of # (utf8mb4_unicode_ci | utf8mb4_0900_ai_ci | utf8mb4_bin) on a # given environment — see _fk_source_col_type docstring above. # We introspect each parent at migration time and match. if not _table_exists(bind, _REQUESTS_TABLE): persons_id = _fk_source_col_type(bind, "identity_persons", "id") integrations_id = _fk_source_col_type(bind, "integrations", "id") dns_records_id = _fk_source_col_type(bind, "cmdb_dns_records", "id") op.create_table( _REQUESTS_TABLE, sa.Column("id", sa.String(36), primary_key=True), sa.Column("requester_person_id", persons_id, nullable=False), sa.Column("action", sa.String(16), nullable=False), sa.Column("integration_id", integrations_id, nullable=False), sa.Column("zone_name", sa.String(253), nullable=False), sa.Column("name", sa.String(256), nullable=False), sa.Column("record_type", sa.String(16), nullable=False), sa.Column("value", sa.Text, nullable=False), sa.Column("ttl", sa.Integer, nullable=False, server_default=sa.text("600")), sa.Column("line", sa.String(64), nullable=True), sa.Column("target_record_id", dns_records_id, nullable=True), sa.Column("purpose", sa.Text, nullable=False), sa.Column("attachment_url", sa.String(1024), nullable=True), sa.Column( "status", sa.String(32), nullable=False, server_default=sa.text("'pending'"), ), sa.Column("feishu_instance_code", sa.String(128), nullable=True), sa.Column("approval_request_id", sa.String(36), nullable=True), sa.Column("executed_record_pk", dns_records_id, nullable=True), sa.Column("error_message", sa.Text, nullable=True), sa.Column("submitted_at", sa.DateTime(timezone=True), nullable=True), sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), sa.Column("executed_at", sa.DateTime(timezone=True), 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(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["requester_person_id"], ["identity_persons.id"], name="fk_dns_req_requester", ), sa.ForeignKeyConstraint( ["integration_id"], ["integrations.id"], name="fk_dns_req_integration", ), sa.ForeignKeyConstraint( ["target_record_id"], ["cmdb_dns_records.id"], name="fk_dns_req_target", ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["executed_record_pk"], ["cmdb_dns_records.id"], name="fk_dns_req_executed", ondelete="SET NULL", ), sa.UniqueConstraint( "feishu_instance_code", name="uq_dns_req_feishu_instance", ), sa.Index("ix_dns_req_requester", "requester_person_id", "status"), sa.Index("ix_dns_req_zone", "zone_name", "status"), sa.Index("ix_dns_req_status_created", "status", "created_at"), mysql_engine="InnoDB", mysql_charset="utf8mb4", ) # ─── (2) cmdb_dns_records.last_owner_review_at ───────────────── if _table_exists(bind, _RECORDS_TABLE) and not _column_exists( bind, _RECORDS_TABLE, _NEW_RECORDS_COL, ): bind.execute(sa.text( f"ALTER TABLE {_RECORDS_TABLE} " f"ADD COLUMN {_NEW_RECORDS_COL} DATETIME(6) NULL" )) # Index supports the recertification sweep: # WHERE deleted_at IS NULL AND owner_person_id IS NOT NULL # AND (last_owner_review_at IS NULL OR last_owner_review_at < ?) if _table_exists(bind, _RECORDS_TABLE) and not _index_exists( bind, _RECORDS_TABLE, "ix_cmdb_dns_records_owner_review", ): bind.execute(sa.text( f"CREATE INDEX ix_cmdb_dns_records_owner_review " f"ON {_RECORDS_TABLE} (owner_person_id, last_owner_review_at)" )) # ─── (3) scheduled jobs ──────────────────────────────────────── _ensure_scheduled_job( bind, name=_JOB_RECERTIFY, display_name="DNS Owner 季度保鲜", description=( "每天 09:30 (Asia/Shanghai) 扫描所有 DNS 记录,若 owner 距上次" "确认已超过 cmdb.dns.recertify_period_days(默认 90 天),按" " owner 聚合发送一条飞书卡片提醒,卡片含 '全部仍在使用' 一键" "续期按钮。NULL 视作从未确认,首次运行会全员通知一次。" ), cron_expr="30 9 * * *", handler_key="cmdb.dns_owner_recertify", config={"recertify_period_days": _SETTING_DEFAULT}, ) _ensure_scheduled_job( bind, name=_JOB_LEAVER, display_name="DNS 离职前转移提醒", description=( "每天 09:30 (Asia/Shanghai) 扫描所有 last_working_day 在未来 7 " "天内的 person,若其名下持有 DNS 记录或域名,飞书通知本人" "尽快走自助流程 delete 或转交,未在离职日处理的将由 T+0 " "force-transfer 兜底转给 manager 链。" ), cron_expr="30 9 * * *", handler_key="cmdb.dns_pre_leaver_handover", config={"warn_days": 7}, ) # ─── (4) platform setting default ────────────────────────────── if _table_exists(bind, "platform_settings"): exists = bind.execute( sa.text( "SELECT 1 FROM platform_settings WHERE `key` = :k LIMIT 1" ), {"k": _SETTING_KEY}, ).scalar() if not exists: bind.execute( sa.text( "INSERT INTO platform_settings " "(`key`, value, description, created_at, updated_at) " "VALUES (:k, :v, :d, NOW(), NOW())" ), { "k": _SETTING_KEY, "v": json.dumps(_SETTING_DEFAULT), "d": ( "DNS owner recertification cadence in days. " "Owners who have not confirmed in this many days " "get a Feishu nag from cmdb.dns_owner_recertify." ), }, ) def downgrade() -> None: bind = op.get_bind() # Drop scheduled jobs (best-effort). if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE name IN (:a, :b)"), {"a": _JOB_RECERTIFY, "b": _JOB_LEAVER}, ) # Drop platform setting (best-effort). if _table_exists(bind, "platform_settings"): bind.execute( sa.text("DELETE FROM platform_settings WHERE `key` = :k"), {"k": _SETTING_KEY}, ) # Drop the requests table. if _table_exists(bind, _REQUESTS_TABLE): bind.execute(sa.text(f"DROP TABLE {_REQUESTS_TABLE}")) # Drop the column + index on cmdb_dns_records. if _table_exists(bind, _RECORDS_TABLE): if _index_exists(bind, _RECORDS_TABLE, "ix_cmdb_dns_records_owner_review"): bind.execute(sa.text( f"DROP INDEX ix_cmdb_dns_records_owner_review " f"ON {_RECORDS_TABLE}" )) if _column_exists(bind, _RECORDS_TABLE, _NEW_RECORDS_COL): bind.execute(sa.text( f"ALTER TABLE {_RECORDS_TABLE} DROP COLUMN {_NEW_RECORDS_COL}" ))