#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import subprocess import sys from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[3] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from scripts.connected_mail_oauth import connected_mail_oauth_env # noqa: E402 from common import ( DeployError, checkout_resolved_ref, clone_or_update_repo, ensure_runtime_mutable_dirs, log, named_credential_master_key_from_env, normalize_runtime_mutable_env, require_commands, resolve_pack_repo_commit_runtime_refs, resolve_repo_ref, ) from deployment_runtime_state import runtime_paths from deployment_plan import ( RepoOperations, repository_urls, resolve_prepare_refs as resolve_prepare_refs_from_plan, validate_database_request, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--env-root", required=True) parser.add_argument("--downstream-ref", required=True) parser.add_argument("--upstream-ref", required=True) parser.add_argument("--official-app-packs-ref", required=True) parser.add_argument("--tidb-database", required=True) parser.add_argument("--backend-port", default="3103") parser.add_argument("--public-base-url", default="http://127.0.0.1:3103") parser.add_argument("--server-name", default="127.0.0.1") parser.add_argument("--install-systemd", dest="install_systemd", action="store_true") parser.add_argument("--no-install-systemd", dest="install_systemd", action="store_false") parser.add_argument("--install-nginx", dest="install_nginx", action="store_true") parser.add_argument("--no-install-nginx", dest="install_nginx", action="store_false") parser.add_argument("--reset-db", dest="reset_db", action="store_true") parser.add_argument("--no-reset-db", dest="reset_db", action="store_false") parser.set_defaults(install_systemd=False, install_nginx=False, reset_db=False) return parser.parse_args() def require_process_env(keys: list[str]) -> dict[str, str]: values: dict[str, str] = {} missing: list[str] = [] for key in keys: value = os.environ.get(key, "").strip() if not value: missing.append(key) continue values[key] = value if missing: raise DeployError(f"missing required env keys: {', '.join(missing)}") return values def optional_process_env(keys: list[str]) -> dict[str, str]: values: dict[str, str] = {} for key in keys: value = os.environ.get(key, "") if value: values[key] = value return values def write_env_file(path: Path, values: dict[str, str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) lines = [f"{key}={value}" for key, value in values.items()] path.write_text("\n".join(lines) + "\n") def build_deploy_id() -> str: stamp = subprocess.run( ["date", "-u", "+%Y%m%dT%H%M%SZ"], capture_output=True, text=True, check=True, ).stdout.strip() return f"exp-{stamp}" def resolve_prepare_refs( *, repos_root: Path, downstream_root: Path, upstream_root: Path, app_packs_root: Path, downstream_ref: str, upstream_ref: str, official_app_packs_ref: str, gh_token: str, ) -> dict[str, str]: urls = repository_urls(error_type=DeployError) resolved_refs = resolve_prepare_refs_from_plan( environment="exp", repos_root=repos_root, downstream_root=downstream_root, upstream_root=upstream_root, app_packs_root=app_packs_root, downstream_repo_url=urls["downstream"], upstream_repo_url=urls["upstream"], official_app_packs_repo_url=urls["official_app_packs"], downstream_ref=downstream_ref, upstream_ref=upstream_ref, official_app_packs_ref=official_app_packs_ref, gh_token=gh_token, operations=RepoOperations( require_commands=require_commands, clone_or_update_repo=clone_or_update_repo, resolve_repo_ref=resolve_repo_ref, checkout_resolved_ref=checkout_resolved_ref, resolve_pack_repo_commit_runtime_refs=resolve_pack_repo_commit_runtime_refs, error_type=DeployError, ), ) log(f"prepare resolved downstream ref: {downstream_ref} -> {resolved_refs['DOWNSTREAM_REF']}") log(f"prepare resolved upstream ref: {upstream_ref} -> {resolved_refs['UPSTREAM_REF']}") log( "prepare resolved official app packs ref: " f"{official_app_packs_ref} -> {resolved_refs['OFFICIAL_APP_PACKS_REF']}" ) return resolved_refs def write_github_output(payload: dict[str, str]) -> None: output_path = os.environ.get("GITHUB_OUTPUT", "").strip() if not output_path: return with Path(output_path).open("a", encoding="utf-8") as handle: for key, value in payload.items(): handle.write(f"{key}={value}\n") def main() -> int: args = parse_args() connected_mail_oauth = connected_mail_oauth_env( os.environ, error_type=DeployError, ) if not args.tidb_database.replace("_", "").isalnum(): raise DeployError(f"unsafe TIDB_DATABASE value: {args.tidb_database}") required = require_process_env( [ "TIDB_HOST", "TIDB_PORT", "TIDB_USER", "GH_TOKEN", "JWT_SECRET", "NEXUS_ADMIN_PASSWORD", ] ) if args.install_systemd: required.update(require_process_env(["SYSTEMD_USER"])) validate_database_request(environment="exp", reset_db=bool(args.reset_db), error_type=DeployError) env_root = Path(args.env_root).expanduser().resolve() paths = runtime_paths(env_root) urls = repository_urls(error_type=DeployError) resolved_refs = resolve_prepare_refs( repos_root=paths["repos_root"], downstream_root=paths["downstream_root"], upstream_root=paths["upstream_root"], app_packs_root=paths["app_packs_root"], downstream_ref=args.downstream_ref, upstream_ref=args.upstream_ref, official_app_packs_ref=args.official_app_packs_ref, gh_token=required["GH_TOKEN"], ) deploy_id = build_deploy_id() deploy_artifact_dir = env_root / "artifacts" / "deploys" / deploy_id runtime_env = { "ENV_NAME": "exp", "ENV_ROOT": str(env_root), "DOWNSTREAM_REPO_URL": urls["downstream"], "DOWNSTREAM_REF": resolved_refs["DOWNSTREAM_REF"], "UPSTREAM_REPO_URL": urls["upstream"], "UPSTREAM_REF": resolved_refs["UPSTREAM_REF"], "OFFICIAL_APP_PACKS_REPO_URL": urls["official_app_packs"], "OFFICIAL_APP_PACKS_REF": resolved_refs["OFFICIAL_APP_PACKS_REF"], "DEPLOYMENT_AUTHORITY_MODE": resolved_refs.get("DEPLOYMENT_AUTHORITY_MODE", "explicit_debug"), "RELEASE_REFS_SOURCE": resolved_refs.get("RELEASE_REFS_SOURCE", "explicit"), "RELEASE_CLASS": resolved_refs.get("RELEASE_CLASS", "false"), "BACKEND_PORT": args.backend_port, "PUBLIC_BASE_URL": args.public_base_url, "SERVER_NAME": args.server_name, "TIDB_HOST": required["TIDB_HOST"], "TIDB_PORT": required["TIDB_PORT"], "TIDB_USER": required["TIDB_USER"], "TIDB_DATABASE": args.tidb_database, "JWT_SECRET": required["JWT_SECRET"], "NEXUS_ADMIN_PASSWORD": required["NEXUS_ADMIN_PASSWORD"], "NEXUSCRM_NC_MASTER_KEY": named_credential_master_key_from_env(), "GH_TOKEN": required["GH_TOKEN"], "INSTALL_SYSTEMD": "1" if args.install_systemd else "0", "INSTALL_NGINX": "1" if args.install_nginx else "0", "RESET_DB": "1" if args.reset_db else "0", "RESET_TIDB_BRANCH": "0", "BACKEND_GO_BUILD_TAGS": "crm", # EXP opts in to the guarded L0 public OAuth PKCE gate. Envs whose # prepare script does not set this key remain fail-closed. "NEXUSCRM_PUBLIC_OAUTH_PKCE_ENABLED": "true", "DEPLOY_ID": deploy_id, "DEPLOY_ARTIFACT_DIR": str(deploy_artifact_dir), } if args.install_systemd: runtime_env["SYSTEMD_USER"] = required["SYSTEMD_USER"] runtime_env.update( optional_process_env( [ "TIDB_PASSWORD", "TIDB_TLS", "GOOGLE_CLIENT_ID", "GOOGLE_CLIENT_SECRET", "REACT_APP_GOOGLE_CLIENT_ID", "NEXUS_LOGIN_METHODS", "NEXUS_API_HOURLY_LIMIT", "NEXUS_API_DAILY_LIMIT", "EXP_HUBSPOT_APP_NAME", "EXP_HUBSPOT_APP_ID", "EXP_HUBSPOT_CLIENT_ID", "EXP_HUBSPOT_PROJECT_ID", "EXP_HUBSPOT_AUTH_TYPE", "EXP_HUBSPOT_SUBSCRIPTION_COUNT", "EXP_HUBSPOT_WEBHOOK_URL", "EXP_HUBSPOT_CLIENT_SECRET", "EXP_HUBSPOT_ACCESS_TOKEN", "EXP_HUBSPOT_ACTION_OPTIONS_TOKEN", "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT", "FILE_STORAGE_PROVIDER", "FILE_STORAGE_ROOT", "FILE_STORAGE_LEGACY_UPLOAD_ROOT", "AUDIT_FINAL_PACK", "SKIP_BACKEND_START", "TMPDIR", "GOCACHE", "GOMODCACHE", "GOTMPDIR", "CGO_ENABLED", "CC", "CXX", "SKIP_ENSURE_INDEXES", "NEXUSCRM_REQUEST_CAPTURE", "NEXUSCRM_REQUEST_CAPTURE_RESPONSE", ] ) ) runtime_env.update(connected_mail_oauth) runtime_env = normalize_runtime_mutable_env(runtime_env, env_root) ensure_runtime_mutable_dirs(runtime_env) write_env_file(paths["runtime_env_file"], runtime_env) deploy_artifact_dir.mkdir(parents=True, exist_ok=True) payload = { "runtime_env_file": str(paths["runtime_env_file"]), "deploy_artifact_dir": str(deploy_artifact_dir), "downstream_ref": resolved_refs["DOWNSTREAM_REF"], "upstream_ref": resolved_refs["UPSTREAM_REF"], "official_app_packs_ref": resolved_refs["OFFICIAL_APP_PACKS_REF"], "tidb_database": args.tidb_database, "reset_db": runtime_env["RESET_DB"], "public_oauth_pkce_enabled": runtime_env["NEXUSCRM_PUBLIC_OAUTH_PKCE_ENABLED"], } write_github_output(payload) print(json.dumps(payload, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except DeployError as exc: log(f"failed: {exc}") raise SystemExit(1)