#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import shlex import shutil import sys REPO_ROOT = Path(__file__).resolve().parents[2] DEPLOYMENT_CONFIG = REPO_ROOT / "deployments" / "deployment-config.json" PROD_ENV_ROOT = Path("/home/ubuntu/nexuscrm-deployment/online/envs/multi_prod") PROD_RUNTIME_ENV = PROD_ENV_ROOT / "current" / "env" / "runtime.env" SUPPORTED_TARGETS = ("test1", "test2", "test3", "staging", "prod") sys.path.insert(0, str(REPO_ROOT / "deploy" / "online" / "runtime")) import common as prod_common # noqa: E402 from prepare_multi_prod_runtime_env import ( # noqa: E402 EXPECTED_PROD_TIDB_DATABASE, EXPECTED_PROD_TIDB_PORT, PROD_RESET_NO_ACTIVE_RELEASE, capture_prod_reset_source_identity, update_env_file, validate_prod_tidb_binding, ) sys.path.pop(0) sys.modules.pop("common", None) sys.modules.pop("hubspot_prod_contract", None) sys.path.insert(0, str(REPO_ROOT / "deployments" / "offline" / "runtime")) from common import load_env_file # noqa: E402 sys.path.insert(0, str(REPO_ROOT / "scripts" / "salesforce_migration")) from inspect_salesforce_opportunity_stage_history import ( # noqa: E402 DbConfig as StageDbConfig, opportunity_stage_target_fingerprint, ) from migrate_salesforce_field_history import ( # noqa: E402 DbConfig as FieldDbConfig, target_connection_fingerprint, ) class TargetError(RuntimeError): pass def validate_prod_active_release() -> None: try: identity = capture_prod_reset_source_identity(PROD_ENV_ROOT) except prod_common.DeployError as exc: raise TargetError(f"active prod release identity is invalid: {exc}") from exc if identity["release_tag"] == PROD_RESET_NO_ACTIVE_RELEASE: raise TargetError("prod has no active release") def resolve_staging_runtime() -> tuple[Path, str]: from prepare_staging_runtime_env import ( # noqa: PLC0415 infer_active_d3_slot_from_live_conf, runtime_env_file, ) slot = infer_active_d3_slot_from_live_conf("staging") return runtime_env_file("staging", slot), slot def resolve_runtime_source(target: str) -> tuple[Path, str]: if target in {"test1", "test2", "test3"}: return ( Path( f"/home/ec2-user/nexuscrm-deployment/offline/envs/test/{target}/env/runtime.env" ), target, ) if target == "staging": return resolve_staging_runtime() if target == "prod": validate_prod_active_release() return PROD_RUNTIME_ENV, "active" raise TargetError(f"unsupported migration target: {target}") def expected_non_prod_values(target: str, resolved_slot: str) -> dict[str, str]: config = json.loads(DEPLOYMENT_CONFIG.read_text(encoding="utf-8")) environment = "test" if target.startswith("test") else "staging" topology = config["environments"][environment]["topology"] if resolved_slot not in topology["targets"]: raise TargetError("selected runtime slot is not configured") tidb = topology["target_config"][resolved_slot]["tidb"] return { "TIDB_HOST": str(tidb["host"]), "TIDB_PORT": str(tidb["port"]), "TIDB_USER": str(tidb["user"]), "TIDB_DATABASE": "nexuscrm", "TIDB_CLOUD_BRANCH_ID": str(tidb["branch_id"]), } def apply_prod_migration_identity( runtime_copy: Path, runtime: dict[str, str], ) -> None: updates = { "TIDB_USER": os.environ.get("MULTI_PROD_TIDB_USER", "").strip(), "TIDB_PASSWORD": os.environ.get("MULTI_PROD_TIDB_PASSWORD", ""), } missing = [key for key, value in updates.items() if not value] if missing: raise TargetError( "selected prod environment has no approved TiDB migration identity" ) runtime.update(updates) update_env_file( runtime_copy, {key: shlex.quote(value) for key, value in updates.items()}, ) def expected_prod_values(runtime: dict[str, str]) -> dict[str, str]: try: validate_prod_tidb_binding(runtime) except prod_common.DeployError as exc: raise TargetError(f"selected prod TiDB identity is invalid: {exc}") from exc return { "TIDB_HOST": runtime.get("TIDB_HOST", "").strip(), "TIDB_PORT": EXPECTED_PROD_TIDB_PORT, "TIDB_USER": runtime.get("TIDB_USER", "").strip(), "TIDB_DATABASE": EXPECTED_PROD_TIDB_DATABASE, } def validate_runtime( target: str, resolved_slot: str, runtime: dict[str, str], expected: dict[str, str], ) -> None: expected_env_name = ( f"test-{target}" if target.startswith("test") else f"staging-{resolved_slot}" if target == "staging" else "multi-prod" ) if runtime.get("ENV_NAME") != expected_env_name: raise TargetError("selected runtime does not match the requested environment") for key, value in expected.items(): if not value: raise TargetError(f"selected runtime has no {key}") if runtime.get(key, "") != value: raise TargetError(f"selected runtime {key} does not match its environment") if not runtime.get("TIDB_PASSWORD", "").strip(): raise TargetError("selected runtime has no TiDB credential") def target_fingerprint(scope: str, expected: dict[str, str]) -> str: if scope == "field-history": return target_connection_fingerprint( FieldDbConfig( expected["TIDB_HOST"], int(expected["TIDB_PORT"]), "", "", expected["TIDB_DATABASE"], None, False, ) ) if scope == "opportunity-stage-history": return opportunity_stage_target_fingerprint( StageDbConfig( expected["TIDB_HOST"], int(expected["TIDB_PORT"]), expected["TIDB_USER"], "", expected["TIDB_DATABASE"], None, False, ) ) raise TargetError(f"unsupported migration scope: {scope}") def prepare_target( target: str, scope: str, runtime_copy: Path, provenance_output: Path, ) -> None: source, resolved_slot = resolve_runtime_source(target) if not source.is_file(): raise TargetError("selected runtime env is unavailable") runtime_copy.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source, runtime_copy) runtime_copy.chmod(0o600) runtime = load_env_file(runtime_copy) if target == "prod": apply_prod_migration_identity(runtime_copy, runtime) expected = ( expected_prod_values(runtime) if target == "prod" else expected_non_prod_values(target, resolved_slot) ) validate_runtime(target, resolved_slot, runtime, expected) payload = { "schema_version": "pingcap-salesforce-history-target-provenance/v1", "selected_environment": target, "resolved_runtime_slot": resolved_slot, "deploy_id": runtime.get("DEPLOY_ID", ""), "deployed_downstream_ref": runtime.get("DOWNSTREAM_REF", ""), "deployed_upstream_ref": runtime.get("UPSTREAM_REF", ""), "deployed_official_app_packs_ref": runtime.get( "OFFICIAL_APP_PACKS_REF", "" ), "target_connection_fingerprint": target_fingerprint(scope, expected), "credential_namespace_sha256": hashlib.sha256( expected["TIDB_USER"].split(".", 1)[0].encode() ).hexdigest(), } branch_id = expected.get("TIDB_CLOUD_BRANCH_ID", "") if branch_id: payload["runtime_branch_id_sha256"] = hashlib.sha256( branch_id.encode() ).hexdigest() provenance_output.write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--target", choices=SUPPORTED_TARGETS, required=True) parser.add_argument( "--scope", choices=("field-history", "opportunity-stage-history"), required=True, ) parser.add_argument("--runtime-copy", type=Path, required=True) parser.add_argument("--provenance-output", type=Path, required=True) return parser.parse_args() def main() -> int: args = parse_args() prepare_target( args.target, args.scope, args.runtime_copy, args.provenance_output, ) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except TargetError as exc: raise SystemExit(f"target resolution failed: {exc}")