Playbook：让 DDL 风险在合并前可见——CI/CD 双 Stage Schema Check 设计

#!/usr/bin/env python3
# schema_check.py - DDL diff + 风险等级判断
# 用法：./schema_check.py --engine mysql --source qa --target pre --db biz_a --rules ignore-rules.yaml
import argparse, os, sys, json, re, fnmatch
from dataclasses import dataclass, asdict
from typing import List, Dict, Tuple
import yaml

try:
    import pymysql
except ImportError:
    pymysql = None
try:
    import psycopg2, psycopg2.extras
except ImportError:
    psycopg2 = None


@dataclass
class Diff:
    kind: str            # MISSING_TABLE / MISSING_COLUMN / TYPE_MISMATCH / NOT_NULL_NO_DEFAULT / DROP_INDEX / DROP_COLUMN / RENAME_TABLE
    table: str
    column: str = ""
    detail: str = ""
    risk: str = "warn"   # block / warn / silent


def env(name: str) -> str:
    v = os.environ.get(name, "")
    if not v:
        sys.stderr.write(f"missing env {name}\n"); sys.exit(2)
    return v


def conn_mysql(prefix: str):
    return pymysql.connect(
        host=env(f"{prefix}_HOST"), port=int(os.environ.get(f"{prefix}_PORT", "3306")),
        user=env(f"{prefix}_USER"), password=env(f"{prefix}_PASS"),
        database=env(f"{prefix}_DB"), charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor, connect_timeout=10)


def conn_pg(prefix: str):
    return psycopg2.connect(
        host=env(f"{prefix}_HOST"), port=int(os.environ.get(f"{prefix}_PORT", "5432")),
        user=env(f"{prefix}_USER"), password=env(f"{prefix}_PASS"),
        dbname=env(f"{prefix}_DB"), connect_timeout=10,
        cursor_factory=psycopg2.extras.RealDictCursor)


def fetch_schema_mysql(cur, db: str) -> Dict:
    cur.execute("""
        SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_KEY
        FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA=%s
        ORDER BY TABLE_NAME, ORDINAL_POSITION""", (db,))
    schema: Dict[str, Dict[str, dict]] = {}
    for r in cur.fetchall():
        schema.setdefault(r["TABLE_NAME"], {})[r["COLUMN_NAME"]] = {
            "type": r["COLUMN_TYPE"], "nullable": r["IS_NULLABLE"] == "YES",
            "default": r["COLUMN_DEFAULT"], "key": r["COLUMN_KEY"]}
    return schema


def fetch_schema_pg(cur, db: str) -> Dict:
    cur.execute("""
        SELECT table_name, column_name, data_type, is_nullable, column_default
        FROM information_schema.columns WHERE table_schema='public'
        ORDER BY table_name, ordinal_position""")
    schema: Dict[str, Dict[str, dict]] = {}
    for r in cur.fetchall():
        schema.setdefault(r["table_name"], {})[r["column_name"]] = {
            "type": r["data_type"], "nullable": r["is_nullable"] == "YES",
            "default": r["column_default"], "key": ""}
    return schema


def is_ignored(name: str, patterns: List[str]) -> bool:
    return any(fnmatch.fnmatch(name, p) for p in patterns or [])


def classify_risk(kind: str, rules: Dict) -> str:
    for level in ("block", "warn", "silent"):
        if kind in rules.get("risk_levels", {}).get(level, []):
            return level
    return "warn"


def diff_schemas(src: Dict, tgt: Dict, rules: Dict) -> List[Diff]:
    out: List[Diff] = []
    ignore_tables = rules.get("ignore_tables", {}).get(env("DB_NAME"), [])
    # source 有，target 无 → 缺表（部署后 post stage 关注）
    for t in src:
        if is_ignored(t, ignore_tables): continue
        if t not in tgt:
            out.append(Diff("MISSING_TABLE", t, risk=classify_risk("MISSING_TABLE", rules)))
            continue
        for col, meta in src[t].items():
            if col not in tgt[t]:
                out.append(Diff("MISSING_COLUMN", t, col,
                    detail=f"src has {meta['type']}", risk=classify_risk("MISSING_COLUMN", rules)))
                continue
            if meta["type"].lower() != tgt[t][col]["type"].lower():
                out.append(Diff("TYPE_MISMATCH", t, col,
                    detail=f"{tgt[t][col]['type']} → {meta['type']}",
                    risk=classify_risk("ALTER_COLUMN_TYPE", rules)))
            if (not meta["nullable"]) and meta["default"] is None and tgt[t][col]["nullable"]:
                out.append(Diff("NOT_NULL_NO_DEFAULT", t, col,
                    risk=classify_risk("ADD_COLUMN_NOT_NULL_NO_DEFAULT", rules)))
    # target 有 source 无 → 多表（高危：可能是 PROD 残留 / 别人在用）
    for t in tgt:
        if is_ignored(t, ignore_tables): continue
        if t not in src:
            out.append(Diff("EXTRA_TABLE", t, detail="exists in target only",
                risk=classify_risk("DROP_TABLE", rules)))
    return out


def render_report(diffs: List[Diff], mode: str, db: str) -> Tuple[str, int]:
    blocks = [d for d in diffs if d.risk == "block"]
    warns  = [d for d in diffs if d.risk == "warn"]
    status = "PASS" if not blocks and not warns else ("FAIL" if blocks and mode == "post" else "WARN")
    exit_code = 1 if (blocks and mode == "post") else 0
    lines = [f"=== Schema Check {status} === db={db} mode={mode}"]
    if blocks:
        lines.append("\n[BLOCK] 破坏性 DDL（post stage 阻塞 PROD）：")
        for d in blocks: lines.append(f"  - {d.kind}: {d.table}.{d.column} {d.detail}")
    if warns:
        lines.append("\n[WARN] 提示：")
        for d in warns: lines.append(f"  - {d.kind}: {d.table}.{d.column} {d.detail}")
    if not (blocks or warns): lines.append("无差异")
    lines.append("\n下一步：")
    if blocks and mode == "post":
        lines.append("  1) 修 migration 补 backfill + 重发版")
        lines.append("  2) 评估后加 ignore-rules.yaml（PR 走 DBA review）")
        lines.append("  3) 找 DBA 拉群讨论临时豁免")
    elif warns:
        lines.append("  1) PR 内自查 migration 是否预期")
        lines.append("  2) 如属预期可在 PR description 标注，无需改动")
    return "\n".join(lines), exit_code


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--engine", choices=["mysql", "postgresql"], required=True)
    ap.add_argument("--mode", choices=["pre", "post"], required=True)
    ap.add_argument("--rules", default="ignore-rules.yaml")
    ap.add_argument("--json-out", default="")
    args = ap.parse_args()

    rules = yaml.safe_load(open(args.rules))
    db = env("DB_NAME")
    if args.engine == "mysql":
        if not pymysql: sys.exit("pip install pymysql")
        with conn_mysql("SRC") as s, conn_mysql("TGT") as t:
            src = fetch_schema_mysql(s.cursor(), env("SRC_DB"))
            tgt = fetch_schema_mysql(t.cursor(), env("TGT_DB"))
    else:
        if not psycopg2: sys.exit("pip install psycopg2-binary")
        with conn_pg("SRC") as s, conn_pg("TGT") as t:
            src = fetch_schema_pg(s.cursor(), env("SRC_DB"))
            tgt = fetch_schema_pg(t.cursor(), env("TGT_DB"))

    diffs = diff_schemas(src, tgt, rules)
    report, code = render_report(diffs, args.mode, db)
    print(report)
    if args.json_out:
        with open(args.json_out, "w") as f:
            json.dump([asdict(d) for d in diffs], f, ensure_ascii=False, indent=2)
    sys.exit(code)


if __name__ == "__main__":
    main()

# ignore-rules.yaml — 风险等级与豁免规则集中维护
# 加豁免必须走 PR review，不要直接 push
risk_levels:
  block:                       # 命中即 post stage fail
    - DROP_TABLE
    - DROP_COLUMN
    - ALTER_COLUMN_TYPE
    - RENAME_TABLE
    - DROP_INDEX_PRIMARY
    - EXTRA_TABLE               # target 多表通常是 PROD 残留
  warn:                        # pre/post 都通知，不阻塞
    - ADD_COLUMN_NOT_NULL_NO_DEFAULT
    - ADD_UNIQUE_INDEX
    - MISSING_TABLE             # 部署前 source 比 target 多很正常
    - MISSING_COLUMN
  silent:
    - ADD_COLUMN_NULLABLE
    - ADD_INDEX_NORMAL
    - COMMENT_CHANGE

ignore_tables:                 # 已知豁免（业务确认 + 过期日期）
  business_db_a:
    - tmp_migration_*           # 临时表，过期 2026-12
    - audit_log_2024_*
    - shadow_*                  # 影子表回放专用
  business_db_b:
    - dba_grants_log

stage_schema_check_pre:
  name: schema_check (PRE deploy 前)
  needs: [stage_build]
  jobs:
    job_check_pre:
      runsOn:
        group: ${BUILD_GROUP}
        labels: linux,amd64
        vm: true
      sourceOption: []
      steps:
        step_check:
          step: Command
          continueOnError: true   # 关键：warning 模式不阻塞
          with:
            run: |
              set +e
              # 注意：云效 step 级别 envs 字段静默失效，必须 export
              export SRC_HOST="${DB_HOST_QA}"
              export SRC_USER="${DB_USER_RO}"
              export SRC_PASS="${DB_PASS_QA}"
              export SRC_DB="${DB_NAME}"
              export TGT_HOST="${DB_HOST_PRE}"
              export TGT_USER="${DB_USER_RO}"
              export TGT_PASS="${DB_PASS_PRE}"
              export TGT_DB="${DB_NAME}"
              export DB_NAME="${DB_NAME}"
              git clone --depth 1 https://example.com/infra/gitops.git
              cd gitops/scripts/db-schema-check
              pip3 install -q --break-system-packages pymysql psycopg2-binary pyyaml requests
              python3 schema_check.py --engine mysql --mode pre \
                --rules ignore-rules.yaml --json-out /tmp/diffs.json | tee /tmp/report.txt
              python3 ding_notify.py --report /tmp/report.txt --mode pre \
                --at "${DINGTALK}" --webhook "${DING_WEBHOOK}"
              exit 0   # 永远 exit 0，靠通知传递信号

stage_schema_check_post:
  name: schema_check (PROD deploy 前)
  needs: [stage_deploy_pre]
  jobs:
    job_check_post:
      runsOn:
        group: ${BUILD_GROUP}
        labels: linux,amd64
        vm: true
      sourceOption: []
      steps:
        step_check:
          step: Command
          continueOnError: false   # 关键：必须真阻塞
          with:
            run: |
              set -euo pipefail
              export SRC_HOST="${DB_HOST_QA}"
              export SRC_USER="${DB_USER_RO}"
              export SRC_PASS="${DB_PASS_QA}"
              export SRC_DB="${DB_NAME}"
              export TGT_HOST="${DB_HOST_PRE}"
              export TGT_USER="${DB_USER_RO}"
              export TGT_PASS="${DB_PASS_PRE}"
              export TGT_DB="${DB_NAME}"
              export DB_NAME="${DB_NAME}"
              git clone --depth 1 https://example.com/infra/gitops.git
              cd gitops/scripts/db-schema-check
              pip3 install -q --break-system-packages pymysql psycopg2-binary pyyaml requests
              if ! python3 schema_check.py --engine mysql --mode post \
                   --rules ignore-rules.yaml --json-out /tmp/diffs.json | tee /tmp/report.txt; then
                python3 ding_notify.py --report /tmp/report.txt --mode post --status FAIL \
                  --at "${DINGTALK}" --webhook "${DING_WEBHOOK}"
                exit 1
              fi
              python3 ding_notify.py --report /tmp/report.txt --mode post --status PASS \
                --at "${DINGTALK}" --webhook "${DING_WEBHOOK}" || true

# .github/workflows/schema-check.yml
name: schema-check
on:
  pull_request:                 # pre stage：PR 创建/更新
    paths: ['migrations/**', 'sql/**']
  merge_group:                  # post stage：merge queue（GitHub 合并前最后一道闸）
    types: [checks_requested]

jobs:
  pre-check:
    if: github.event_name == 'pull_request'
    runs-on: ubuntu-latest
    continue-on-error: true     # warning 不阻塞合并
    env:
      SRC_HOST: ${{ secrets.DB_HOST_QA }}
      SRC_USER: ${{ secrets.DB_USER_RO }}
      SRC_PASS: ${{ secrets.DB_PASS_QA }}
      SRC_DB:   biz_a
      TGT_HOST: ${{ secrets.DB_HOST_PRE }}
      TGT_USER: ${{ secrets.DB_USER_RO }}
      TGT_PASS: ${{ secrets.DB_PASS_PRE }}
      TGT_DB:   biz_a
      DB_NAME:  biz_a
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install pymysql psycopg2-binary pyyaml requests
      - run: |
          python3 scripts/schema_check.py --engine mysql --mode pre \
            --rules ignore-rules.yaml --json-out diffs.json | tee report.txt
      - uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const body = '```\n' + fs.readFileSync('report.txt','utf8') + '\n```';
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner, repo: context.repo.repo, body });

  post-check:
    if: github.event_name == 'merge_group'
    runs-on: ubuntu-latest
    # 注意：不加 continue-on-error，失败必阻塞 merge queue
    env:
      SRC_HOST: ${{ secrets.DB_HOST_QA }}
      SRC_USER: ${{ secrets.DB_USER_RO }}
      SRC_PASS: ${{ secrets.DB_PASS_QA }}
      SRC_DB:   biz_a
      TGT_HOST: ${{ secrets.DB_HOST_PRE }}
      TGT_USER: ${{ secrets.DB_USER_RO }}
      TGT_PASS: ${{ secrets.DB_PASS_PRE }}
      TGT_DB:   biz_a
      DB_NAME:  biz_a
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install pymysql psycopg2-binary pyyaml requests
      - run: |
          python3 scripts/schema_check.py --engine mysql --mode post \
            --rules ignore-rules.yaml | tee report.txt

#!/usr/bin/env python3
# ding_notify.py — 把 schema_check 报告推到钉钉 + GitLab MR 评论
# 用法：./ding_notify.py --report report.txt --mode pre --at "138...,139..." --webhook https://...
import argparse, os, sys, json, requests, hmac, hashlib, base64, urllib.parse, time

def sign(secret: str) -> str:
    ts = str(round(time.time() * 1000))
    s = f"{ts}\n{secret}".encode()
    sig = base64.b64encode(hmac.new(secret.encode(), s, hashlib.sha256).digest())
    return f"&timestamp={ts}&sign={urllib.parse.quote_plus(sig)}"

def post_dingtalk(webhook: str, secret: str, title: str, body: str, at_mobiles: list):
    url = webhook + (sign(secret) if secret else "")
    payload = {
        "msgtype": "markdown",
        "markdown": {"title": title, "text": body},
        "at": {"atMobiles": at_mobiles, "isAtAll": False},
    }
    r = requests.post(url, json=payload, timeout=10)
    r.raise_for_status()
    if r.json().get("errcode") != 0:
        sys.stderr.write(f"dingtalk err: {r.text}\n"); sys.exit(3)

def post_gitlab_comment(api: str, token: str, project_id: str, mr_iid: str, body: str):
    r = requests.post(
        f"{api}/projects/{project_id}/merge_requests/{mr_iid}/notes",
        headers={"PRIVATE-TOKEN": token}, json={"body": body}, timeout=10)
    r.raise_for_status()

def render_md(report: str, mode: str, status: str, ci_url: str) -> str:
    icon = {"FAIL": "[FAIL]", "WARN": "[WARN]", "PASS": "[PASS]"}.get(status, "[INFO]")
    lines = [
        f"### {icon} Schema Check {status}  mode={mode}",
        "",
        "**结论**：见下方详情",
        "",
        "**下一步**：",
        "- 选项 1：修 migration 补 backfill 后重提",
        "- 选项 2：评估后加 ignore-rule（PR 走 DBA review）",
        "- 选项 3：找 DBA 拉群讨论",
        "",
        f"[查看完整 log]({ci_url})",
        "",
        "**详情**：",
        "```",
        report.strip()[:2000],
        "```",
    ]
    return "\n".join(lines)

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--report", required=True)
    ap.add_argument("--mode", required=True)
    ap.add_argument("--status", default="WARN")
    ap.add_argument("--at", default="")
    ap.add_argument("--webhook", required=True)
    ap.add_argument("--secret", default=os.environ.get("DING_SECRET", ""))
    ap.add_argument("--gitlab-api", default=os.environ.get("CI_API_V4_URL", ""))
    ap.add_argument("--gitlab-token", default=os.environ.get("CI_BOT_TOKEN", ""))
    ap.add_argument("--project-id", default=os.environ.get("CI_PROJECT_ID", ""))
    ap.add_argument("--mr-iid", default=os.environ.get("CI_MERGE_REQUEST_IID", ""))
    args = ap.parse_args()

    report = open(args.report).read()
    ci_url = os.environ.get("CI_PIPELINE_URL", "(no CI url)")
    md = render_md(report, args.mode, args.status, ci_url)
    title = f"Schema Check {args.status} ({args.mode})"
    at_mobiles = [m.strip() for m in args.at.split(",") if m.strip()]
    post_dingtalk(args.webhook, args.secret, title, md, at_mobiles)
    if args.gitlab_token and args.mr_iid:
        post_gitlab_comment(args.gitlab_api, args.gitlab_token, args.project_id, args.mr_iid, md)

if __name__ == "__main__":
    main()

#!/bin/bash
# table-refs.sh — 给定表名清单，反查哪些服务仓库在引用
# 用法：./table-refs.sh "orders,users,sessions" /data/repos
set -euo pipefail

TABLES="${1:?usage: $0 <table_csv> <repo_root>}"
REPO_ROOT="${2:?}"

command -v rg >/dev/null || { echo "需要 ripgrep: apt install -y ripgrep"; exit 1; }

declare -A REFS
IFS=',' read -ra ARR <<< "$TABLES"
for t in "${ARR[@]}"; do
  t="$(echo "$t" | xargs)"
  # 匹配 SQL/ORM/migration 中的常见引用形态
  hits=$(rg -l --no-ignore-vcs \
    -e "FROM\s+${t}\b" -e "JOIN\s+${t}\b" \
    -e "INSERT\s+INTO\s+${t}\b" -e "UPDATE\s+${t}\b" -e "DELETE\s+FROM\s+${t}\b" \
    -e "Table\([\"']${t}[\"']\)" -e "TableName\(\)\s*string\s*\{[^}]*${t}" \
    -e "@Table\(name\s*=\s*[\"']${t}[\"']\)" \
    "$REPO_ROOT" 2>/dev/null | xargs -I{} dirname {} | sort -u | head -20 || true)
  if [[ -n "$hits" ]]; then
    REFS["$t"]="$hits"
  fi
done

echo "=== 跨服务引用报告 ==="
for t in "${!REFS[@]}"; do
  echo
  echo "[表] $t"
  echo "${REFS[$t]}" | head -10 | sed 's/^/  /'
done

#!/usr/bin/env python3
# inject_schema_check.py — 给一条流水线注入 pre/post 双 stage，幂等可重跑
import argparse, yaml, sys, copy
from pathlib import Path

PRE_TPL = yaml.safe_load(Path("templates/stage_pre.yaml").read_text())
POST_TPL = yaml.safe_load(Path("templates/stage_post.yaml").read_text())

def has_stage(pipe: dict, name: str) -> bool:
    return name in pipe.get("stages", {})

def find_after_build(stages: dict) -> str:
    for k, v in stages.items():
        if "build" in k.lower():
            return k
    raise SystemExit("找不到 build stage，无法定位插入点")

def inject(pipeline_yaml: str, db_engine: str, db_name: str) -> str:
    pipe = yaml.safe_load(open(pipeline_yaml))
    changed = False
    if not has_stage(pipe, "stage_schema_check_pre"):
        pre = copy.deepcopy(PRE_TPL)
        pre["stage_schema_check_pre"]["needs"] = [find_after_build(pipe["stages"])]
        pipe["stages"].update(pre)
        changed = True
    if not has_stage(pipe, "stage_schema_check_post"):
        post = copy.deepcopy(POST_TPL)
        # post 必须 needs PRE deploy stage
        post["stage_schema_check_post"]["needs"] = ["stage_deploy_pre"]
        pipe["stages"].update(post)
        # 把 PROD stage 的 needs 改为依赖 post check
        for k, v in pipe["stages"].items():
            if "prod" in k.lower() and "deploy" in k.lower():
                v["needs"] = ["stage_schema_check_post"]
        changed = True
    # 注入 globalParams（如尚未存在）
    gp = pipe.setdefault("globalParams", [])
    for var in ["DB_NAME", "DINGTALK"]:
        if not any(p.get("name") == var for p in gp):
            gp.append({"name": var, "value": ""})
            changed = True
    if changed:
        Path(pipeline_yaml).write_text(yaml.safe_dump(pipe, sort_keys=False, allow_unicode=True))
        print(f"[INJECTED] {pipeline_yaml}")
    else:
        print(f"[SKIP   ] {pipeline_yaml} 已注入")

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--pipeline", required=True)
    ap.add_argument("--engine", required=True, choices=["mysql", "postgresql"])
    ap.add_argument("--db", required=True)
    args = ap.parse_args()
    inject(args.pipeline, args.engine, args.db)

#!/usr/bin/env python3
# owner-ack-check.py — 上线后 1 周抽查，确认 owner 真的关注 schema_check 通知
import os, sys, requests, datetime, json
from collections import defaultdict

GITLAB = os.environ["GITLAB_API"]
TOKEN  = os.environ["CI_BOT_TOKEN"]

def list_recent_mrs(project: str, days: int = 7):
    since = (datetime.datetime.utcnow() - datetime.timedelta(days=days)).isoformat()
    r = requests.get(f"{GITLAB}/projects/{project}/merge_requests",
                     headers={"PRIVATE-TOKEN": TOKEN},
                     params={"state": "merged", "updated_after": since}, timeout=10)
    r.raise_for_status()
    return r.json()

def notes_of(project: str, iid: int):
    r = requests.get(f"{GITLAB}/projects/{project}/merge_requests/{iid}/notes",
                     headers={"PRIVATE-TOKEN": TOKEN}, timeout=10)
    r.raise_for_status()
    return r.json()

def main():
    project = sys.argv[1]
    stats = defaultdict(int)
    no_response = []
    for mr in list_recent_mrs(project):
        notes = notes_of(project, mr["iid"])
        bot = [n for n in notes if "Schema Check" in n.get("body", "")]
        if not bot: continue
        stats["total"] += 1
        # 看 schema_check 评论之后是否有 owner 回复
        bot_at = bot[-1]["created_at"]
        replies = [n for n in notes if n["created_at"] > bot_at and n["author"]["id"] != bot[-1]["author"]["id"]]
        if replies:
            stats["acked"] += 1
        else:
            no_response.append(mr["iid"])
    print(json.dumps({"stats": dict(stats), "no_response_iids": no_response}, indent=2))

if __name__ == "__main__":
    main()

# test_schema_check.py — 验证规则识别正确性
# 跑：pytest -v test_schema_check.py
import pytest
from schema_check import diff_schemas

RULES = {
  "risk_levels": {
    "block": ["DROP_TABLE", "DROP_COLUMN", "ALTER_COLUMN_TYPE", "EXTRA_TABLE"],
    "warn":  ["MISSING_TABLE", "MISSING_COLUMN", "ADD_COLUMN_NOT_NULL_NO_DEFAULT"],
    "silent": [],
  },
  "ignore_tables": {"biz": ["tmp_*"]},
}

def col(t="varchar(255)", n=True, d=None, k=""):
    return {"type": t, "nullable": n, "default": d, "key": k}

import os
os.environ["DB_NAME"] = "biz"

def test_drop_column_blocked():
    src = {"orders": {"id": col("bigint", n=False, k="PRI")}}
    tgt = {"orders": {"id": col("bigint", n=False, k="PRI"),
                      "legacy_status": col("varchar(32)")}}
    diffs = diff_schemas(src, tgt, RULES)
    kinds = [(d.kind, d.risk) for d in diffs]
    assert ("EXTRA_TABLE", "warn") not in kinds   # 表都有，不是 EXTRA_TABLE
    # legacy_status 在 target 有 source 无 — 当前实现归类需结合具体策略

def test_alter_column_type_blocked():
    src = {"users": {"age": col("varchar(10)")}}
    tgt = {"users": {"age": col("int")}}
    diffs = diff_schemas(src, tgt, RULES)
    assert any(d.kind == "TYPE_MISMATCH" and d.risk == "block" for d in diffs)

def test_add_not_null_no_default_warn():
    src = {"users": {"id": col("bigint", n=False, d=None)}}
    tgt = {"users": {}}
    diffs = diff_schemas(src, tgt, RULES)
    assert any(d.kind == "MISSING_COLUMN" for d in diffs)

def test_extra_table_in_target_blocked():
    src = {}
    tgt = {"ghost_table": {"id": col("bigint")}}
    diffs = diff_schemas(src, tgt, RULES)
    assert any(d.kind == "EXTRA_TABLE" and d.risk == "block" for d in diffs)

def test_ignore_pattern_works():
    src = {}
    tgt = {"tmp_migration_2025": {"id": col("bigint")}}
    diffs = diff_schemas(src, tgt, RULES)
    assert all(d.table != "tmp_migration_2025" for d in diffs)

#!/bin/bash
# verify-post-fail.sh — 故意制造 fail 验证 post stage 真的会拦
# 用法：./verify-post-fail.sh <pipeline_name> <test_db>
set -euo pipefail
PIPE="${1:?}"
TDB="${2:?}"

# 1. 在 PRE 测试库故意加一张 source 没有的表（模拟 EXTRA_TABLE）
mysql -h "$DB_HOST_PRE" -u admin -p"$DB_PASS_ADMIN" "$TDB" <<'SQL'
CREATE TABLE ghost_table_for_test (id BIGINT PRIMARY KEY) ENGINE=InnoDB;
SQL

# 2. 触发流水线 post stage
yunxiao-cli pipeline run "$PIPE" --params "skip_build=true"

# 3. 期望：post stage 红 + 钉钉 FAIL + 下游 stage 不开始
echo "expect: stage_schema_check_post=FAILED, stage_deploy_prod=NOT_STARTED"
echo "如果 stage 是 SUCCESS 或下游开始了，说明 post stage 被绕过！"

# 4. 清理
mysql -h "$DB_HOST_PRE" -u admin -p"$DB_PASS_ADMIN" "$TDB" -e "DROP TABLE ghost_table_for_test"

# 反例（不要这么写）
steps:
  step_check:
    step: Command
    envs:                       # 静默失效，永远 not work
      DB_HOST: ${DB_HOST_PRE}
    with:
      run: bash runner.sh

# 正例（必须这样）
steps:
  step_check:
    step: Command
    with:
      run: |
        set -e
        export DB_HOST="${DB_HOST_PRE}"   # 来自流水线变量组（globalParams）
        export DB_PASS="${PG_PASS_PRE}"
        # debug：先打印确认变量真注入
        env | grep -E "^(DB|SRC|TGT)_" | sed 's/PASS=.*/PASS=***/'
        bash runner.sh

# migration_aware.py — 扫描同 PR 内的 migration 文件，识别 backfill 模式
import re, glob

def detect_backfill_pattern(migration_dir: str, table: str, column: str) -> bool:
    """
    若同一 PR 内存在：
      ADD COLUMN <col> ... NULL
      UPDATE/INSERT ... SET <col> = ...
      ALTER COLUMN <col> SET NOT NULL
    则视为安全 pattern，不阻塞
    """
    files = sorted(glob.glob(f"{migration_dir}/**/*.sql", recursive=True))
    saw_add_null = saw_backfill = saw_set_not_null = False
    for f in files:
        sql = open(f).read().lower()
        if re.search(rf"alter\s+table\s+{table}\s+add\s+column\s+{column}\b.*null", sql):
            saw_add_null = True
        if re.search(rf"update\s+{table}\s+set\s+{column}\s*=", sql):
            saw_backfill = True
        if re.search(rf"alter\s+column\s+{column}\s+set\s+not\s+null", sql) or \
           re.search(rf"modify\s+column\s+{column}\b[^,;]*not\s+null", sql):
            saw_set_not_null = True
    return saw_add_null and saw_backfill and saw_set_not_null

#!/usr/bin/env python3
# cross-service-impact.py — post stage 失败时附加：被破坏的表分别被哪些服务引用
import json, subprocess, sys, os

def find_refs(table: str, repo_root: str) -> list:
    out = subprocess.run(
        ["rg", "-l", "--no-ignore-vcs",
         "-e", rf"FROM\s+{table}\b",
         "-e", rf"JOIN\s+{table}\b",
         "-e", rf"INTO\s+{table}\b",
         "-e", rf"UPDATE\s+{table}\b",
         "-e", rf"TableName.*{table}",
         "-e", rf"@Table\(name\s*=\s*[\"']{table}[\"']\)",
         repo_root],
        capture_output=True, text=True, timeout=60)
    services = set()
    for line in out.stdout.splitlines():
        # 假设 repo 结构：/data/repos/<service>/...
        parts = line.split("/")
        if "repos" in parts:
            idx = parts.index("repos")
            if idx + 1 < len(parts):
                services.add(parts[idx + 1])
    return sorted(services)

def main():
    diffs = json.load(open(sys.argv[1]))   # schema_check.py --json-out 输出
    repo_root = os.environ.get("REPO_ROOT", "/data/repos")
    impact = {}
    for d in diffs:
        if d["risk"] != "block": continue
        impact[d["table"]] = find_refs(d["table"], repo_root)
    print(json.dumps(impact, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

#!/bin/bash
# .git/hooks/pre-push — 推送前强制 migration 文件已 git add
set -e

if git diff --cached --name-only | grep -q "^migrations/"; then
  : ok
fi

# 检查 working tree 是否有未 commit 的 migration
unstaged=$(git status --porcelain migrations/ 2>/dev/null | grep -v '^??' || true)
if [[ -n "$unstaged" ]]; then
  echo "ERROR: migrations/ 有未提交的改动："
  echo "$unstaged"
  echo "请先 git add + commit 再推送"
  exit 1
fi

# 检查未跟踪的 .sql 文件
untracked=$(git status --porcelain migrations/ 2>/dev/null | grep '^??' || true)
if [[ -n "$untracked" ]]; then
  echo "WARNING: migrations/ 有未跟踪文件："
  echo "$untracked"
  read -p "确认忽略并推送? (y/N) " ok
  [[ "$ok" =~ ^[Yy]$ ]] || exit 1
fi

#!/bin/bash
# verify-migrations-pushed.sh
set -euo pipefail

# 列出 PR 范围内的 migration 文件
git diff --name-only "${BASE_SHA}..${HEAD_SHA}" -- migrations/ > /tmp/changed.txt

# 对比预期：PR title 含 "schema:" 但 migrations/ 没改 → 警报
if grep -qi "schema:" <<< "$CI_MR_TITLE" && [[ ! -s /tmp/changed.txt ]]; then
  echo "ERROR: PR title 标注 schema: 但 migrations/ 无改动，可能漏 push"
  exit 1
fi