#!/usr/bin/env python3 """Run one exact tracked Patch Bundle as an independent database operation.""" from __future__ import annotations import argparse import json import re import subprocess import sys from pathlib import Path from typing import Sequence REPO_ROOT = Path(__file__).resolve().parent.parent RUNTIME_DIR = REPO_ROOT / "deployments" / "offline" / "runtime" for entry in (str(REPO_ROOT), str(RUNTIME_DIR)): if entry not in sys.path: sys.path.insert(0, entry) from common import DeployError, load_env_file # noqa: E402 from deployment_patch_replay import ( # noqa: E402 execute_patch_bundle, patch_bundle_members_digest, ) PATCH_BUNDLE_ROOT = Path("docs") / "migration" / "patch-bundles" COMMIT_RE = re.compile(r"[0-9a-f]{40}") def git_capture(repo_root: Path, *args: str) -> str: result = subprocess.run( ["git", "-C", str(repo_root), *args], text=True, capture_output=True, check=False, ) if result.returncode != 0: raise DeployError(f"cannot verify Patch Bundle source with git {' '.join(args)}") return result.stdout.strip() def git_capture_bytes(repo_root: Path, *args: str) -> bytes: result = subprocess.run( ["git", "-C", str(repo_root), *args], capture_output=True, check=False, ) if result.returncode != 0: raise DeployError(f"cannot verify Patch Bundle source with git {' '.join(args)}") return result.stdout def git_patch_bundle_tree_digest( repo_root: Path, source_commit: str, relative: Path, ) -> str: """Digest the reviewed Git blobs, not the mutable working-tree bytes.""" listing = git_capture_bytes( repo_root, "ls-tree", "-r", "-z", "--full-tree", source_commit, "--", relative.as_posix(), ) members: list[tuple[str, bytes]] = [] for record in listing.split(b"\0"): if not record: continue try: metadata, raw_path = record.split(b"\t", 1) mode, object_type, object_id = metadata.split(b" ", 2) tracked_path = Path(raw_path.decode("utf-8")) member_path = tracked_path.relative_to(relative).as_posix() except (UnicodeDecodeError, ValueError) as exc: raise DeployError("reviewed Patch Bundle has an unsafe Git tree entry") from exc if object_type != b"blob" or mode == b"120000" or not member_path: raise DeployError("reviewed Patch Bundle must contain regular tracked files") try: object_name = object_id.decode("ascii") except UnicodeDecodeError as exc: raise DeployError("reviewed Patch Bundle has a malformed Git object") from exc members.append( (member_path, git_capture_bytes(repo_root, "cat-file", "blob", object_name)) ) return patch_bundle_members_digest(members) def require_exact_source_commit(repo_root: Path, expected: str) -> str: expected = str(expected or "").strip() if not COMMIT_RE.fullmatch(expected): raise DeployError("--expected-source-commit must be an exact Git commit") actual = git_capture(repo_root, "rev-parse", "HEAD^{commit}") if actual != expected: raise DeployError( "Patch Bundle checkout does not match --expected-source-commit: " f"{actual} != {expected}" ) return actual def require_patch_request(raw: str) -> Path: requested = str(raw or "").strip() if not requested or any(char in requested for char in ("\0", "\r", "\n", "\\")): raise DeployError("--patch-dir must be one canonical repository-relative path") relative = Path(requested) if relative.is_absolute() or relative.anchor: raise DeployError("--patch-dir must be repository-relative") if any(part in {"", ".", ".."} for part in relative.parts): raise DeployError("--patch-dir must not contain . or ..") try: nested = relative.relative_to(PATCH_BUNDLE_ROOT) except ValueError as exc: raise DeployError( "--patch-dir must stay under docs/migration/patch-bundles" ) from exc if not nested.parts: raise DeployError("--patch-dir must name one Patch Bundle") return relative def resolve_exact_patch_bundle(raw: str, *, repo_root: Path = REPO_ROOT) -> tuple[Path, str]: relative = require_patch_request(raw) root = repo_root.resolve() candidate = root / relative if candidate.is_symlink() or not candidate.is_dir(): raise DeployError(f"Patch Bundle directory is missing or unsafe: {relative}") bundle = candidate.resolve() try: bundle.relative_to((root / PATCH_BUNDLE_ROOT).resolve()) bundle.relative_to(root) except ValueError as exc: raise DeployError("Patch Bundle escaped its tracked source root") from exc dirty = git_capture( root, "status", "--porcelain=v1", "--untracked-files=all", "--", relative.as_posix(), ) if dirty: raise DeployError("Patch Bundle source is modified or untracked") tracked = { Path(line).as_posix() for line in git_capture( root, "ls-tree", "-r", "--name-only", "HEAD", "--", relative.as_posix(), ).splitlines() if line } actual: set[str] = set() for path in bundle.rglob("*"): if path.is_symlink(): raise DeployError(f"Patch Bundle refuses a symlink: {path}") if path.is_file(): actual.add(path.relative_to(root).as_posix()) if not tracked or actual != tracked: raise DeployError("Patch Bundle files do not equal the exact Git tree") return bundle, relative.as_posix() def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Execute one exact tracked Patch Bundle independently from application release." ) ) parser.add_argument("--env-file", required=True, help="Target runtime.env with TiDB authority") parser.add_argument( "--patch-dir", required=True, help="Repository-relative directory under docs/migration/patch-bundles", ) parser.add_argument("--operation-id", required=True, help="Unique migration operation identity") parser.add_argument( "--expected-source-commit", required=True, help="Exact commit containing the reviewed Patch Bundle", ) parser.add_argument( "--evidence-dir", required=True, help="New directory for bounded operation evidence", ) return parser.parse_args(argv) def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) source_commit = require_exact_source_commit( REPO_ROOT, args.expected_source_commit ) patch_dir, source_path = resolve_exact_patch_bundle(args.patch_dir) reviewed_tree_digest = git_patch_bundle_tree_digest( REPO_ROOT, source_commit, Path(source_path) ) env_file = Path(args.env_file).expanduser() if env_file.is_symlink() or not env_file.is_file(): raise DeployError(f"target runtime env is missing or unsafe: {env_file}") envv = load_env_file(env_file.resolve()) evidence_dir = Path(args.evidence_dir).expanduser() if evidence_dir.exists() or evidence_dir.is_symlink(): raise DeployError("--evidence-dir must not already exist") summary = execute_patch_bundle( envv, patch_dir, evidence_dir, operation_id=args.operation_id, source_commit=source_commit, source_path=source_path, expected_tree_digest=reviewed_tree_digest, ) print(json.dumps(summary, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except DeployError as exc: print(f"Patch Bundle operation failed: {exc}", file=sys.stderr) raise SystemExit(1)