#!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path import shutil import subprocess import sys import tempfile from typing import Any REPO_ROOT = Path(__file__).resolve().parents[3] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from scripts.salesforce_migration.file_history_contract import ( # noqa: E402 salesforce_history_archive_authority, ) from common import DeployError, load_env_file from file_recovery_common import ( aws_path, file_sha256, load_hash_bound_json, require_exact_keys, require_sha256, upload_s3_object, verify_s3_object_bytes, write_json_exclusive, ) from file_storage_readiness import ( head_s3_object, load_file_rows, parse_size_bytes, require_destination_within_runtime_s3_authority, require_runtime_s3_authority, validate_logical_s3_key, ) from salesforce_file_projection_snapshot_repair import ( repair_salesforce_file_projection_from_snapshot, snapshot_storage, ) OPERATION = "deal-document-salesforce-history-recovery" REQUIRED_SNAPSHOT_OBJECTS = ("ContentDocumentLink", "ContentVersion") def require_salesforce_history_archive_authority( source_storage: dict[str, Any], *, source_region: str, ) -> dict[str, str]: expected = salesforce_history_archive_authority() actual = { "provider": str(source_storage.get("provider") or "").strip().lower(), "bucket": str(source_storage.get("bucket") or "").strip(), "region": str(source_region or "").strip(), "key_prefix": str(source_storage.get("key_prefix") or "").strip().strip("/"), } if actual != expected: raise DeployError( "Salesforce history recovery source is outside the fixed read-only " f"archive authority: expected={expected!r} actual={actual!r}" ) return expected def bind_runtime_recovery_storage( envv: dict[str, str], destination: dict[str, Any], *, context: str, ) -> dict[str, str]: """Bind one physical destination and derive its durable logical key prefix.""" physical = require_destination_within_runtime_s3_authority( envv, destination, context=context, ) runtime = require_runtime_s3_authority(envv) runtime_prefix = str(runtime["key_prefix"]) physical_prefix = str(physical["key_prefix"]) descendant_prefix = runtime_prefix + "/" if not physical_prefix.startswith(descendant_prefix): raise DeployError( f"{context} must select a logical child below FILE_STORAGE_PREFIX" ) logical_prefix = physical_prefix[len(descendant_prefix) :] if not logical_prefix: raise DeployError(f"{context} logical key prefix is empty") logical_prefix, key_problem = validate_logical_s3_key(logical_prefix) if key_problem: raise DeployError( f"{context} derives an unsafe logical key prefix: {key_problem}" ) return { **physical, "logical_key_prefix": logical_prefix, } def bind_snapshot_input_files( snapshot_dir: Path, snapshot_manifest: dict[str, Any], ) -> list[dict[str, Any]]: objects = snapshot_manifest.get("objects") if not isinstance(objects, list): raise DeployError( "Salesforce history recovery requires a direct snapshot manifest objects array" ) root = snapshot_dir.expanduser().resolve() bindings: list[dict[str, Any]] = [] for object_name in REQUIRED_SNAPSHOT_OBJECTS: matches = [ item for item in objects if isinstance(item, dict) and str(item.get("object") or "").strip() == object_name ] if len(matches) != 1: raise DeployError( f"snapshot manifest must contain exactly one {object_name} object" ) relative_name = str(matches[0].get("file") or "").strip() relative_path = Path(relative_name) if not relative_name or relative_path.is_absolute(): raise DeployError(f"snapshot {object_name} file escapes snapshot directory") source_path = (root / relative_path).resolve() try: source_path.relative_to(root) except ValueError as exc: raise DeployError( f"snapshot {object_name} file escapes snapshot directory" ) from exc if not source_path.is_file(): raise DeployError(f"snapshot {object_name} file is missing") bindings.append( { "object": object_name, "file": relative_path.as_posix(), "size_bytes": source_path.stat().st_size, "sha256": file_sha256(source_path), } ) return bindings def materialize_bound_snapshot( *, source_dir: Path, destination_dir: Path, manifest_name: str, expected_manifest_sha256: str, snapshot_files: list[dict[str, Any]], ) -> None: source_root = source_dir.expanduser().resolve() destination_root = destination_dir.expanduser().resolve() destination_root.mkdir(parents=True, exist_ok=False) manifest_relative = Path(str(manifest_name or "").strip()) if not str(manifest_name or "").strip() or manifest_relative.is_absolute(): raise DeployError("snapshot manifest path escapes snapshot directory") manifest_source = (source_root / manifest_relative).resolve() try: manifest_source.relative_to(source_root) except ValueError as exc: raise DeployError("snapshot manifest path escapes snapshot directory") from exc if file_sha256(manifest_source) != require_sha256( expected_manifest_sha256, "expected snapshot manifest hash", ): raise DeployError("snapshot manifest changed after the bound dry-run") manifest_destination = destination_root / manifest_relative manifest_destination.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(manifest_source, manifest_destination) for binding in snapshot_files: relative_path = Path(str(binding.get("file") or "")) source_path = (source_root / relative_path).resolve() try: source_path.relative_to(source_root) except ValueError as exc: raise DeployError("bound snapshot file escapes snapshot directory") from exc expected_size = int(binding.get("size_bytes") or 0) expected_sha = require_sha256( binding.get("sha256"), f"bound snapshot {binding.get('object')} file hash", ) if ( not source_path.is_file() or source_path.stat().st_size != expected_size or file_sha256(source_path) != expected_sha ): raise DeployError("snapshot metadata changed after the bound dry-run") destination_path = destination_root / relative_path destination_path.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source_path, destination_path) if ( destination_path.stat().st_size != expected_size or file_sha256(destination_path) != expected_sha ): raise DeployError("materialized snapshot metadata hash mismatch") def download_s3_object_to_path( *, bucket: str, key: str, region: str, output_path: Path, aws_cli: str = "aws", ) -> None: command = [ aws_path(aws_cli), "s3api", "get-object", "--bucket", bucket, "--key", key, "--checksum-mode", "ENABLED", ] if region: command.extend(["--region", region]) command.append(str(output_path)) result = subprocess.run(command, text=True, capture_output=True, check=False) if result.returncode != 0: message = (result.stderr or result.stdout or "").strip() raise DeployError( f"failed to read Salesforce recovery S3 object: {message[:1000]}" ) def download_s3_object_digest( *, bucket: str, key: str, region: str, aws_cli: str = "aws", ) -> dict[str, Any]: with tempfile.TemporaryDirectory(prefix="nexuscrm-salesforce-recovery-") as tmp: output_path = Path(tmp) / "object" download_s3_object_to_path( bucket=bucket, key=key, region=region, output_path=output_path, aws_cli=aws_cli, ) return { "size_bytes": output_path.stat().st_size, "sha256": file_sha256(output_path), } def plan_blob_recoveries( planned_files: list[dict[str, Any]], *, source_storage: dict[str, Any], source_region: str, destination_storage: dict[str, Any], destination_region: str, ) -> list[dict[str, Any]]: for label, storage in ( ("snapshot source", source_storage), ("recovery destination", destination_storage), ): if str(storage.get("provider") or "").strip().lower() != "s3": raise DeployError(f"{label} provider must be s3") for key in ("bucket", "key_prefix"): if not str(storage.get(key) or "").strip(): raise DeployError(f"{label} {key} is required") unique_files: dict[str, dict[str, Any]] = {} for planned in planned_files: version_id = str(planned.get("source_content_version_id") or "").strip() if not version_id: raise DeployError("planned Salesforce file lacks SourceContentVersionId") existing = unique_files.get(version_id) if existing and int(existing.get("size_bytes") or 0) != int( planned.get("size_bytes") or 0 ): raise DeployError(f"conflicting planned sizes for ContentVersion {version_id}") unique_files[version_id] = planned source_bucket = str(source_storage["bucket"]).strip() source_prefix = str(source_storage["key_prefix"]).strip().strip("/") destination_bucket = str(destination_storage["bucket"]).strip() destination_prefix = str(destination_storage["key_prefix"]).strip().strip("/") destination_logical_prefix = str( destination_storage.get("logical_key_prefix") or "" ).strip().strip("/") if not destination_logical_prefix: raise DeployError("recovery destination logical_key_prefix is required") result: list[dict[str, Any]] = [] for version_id in sorted(unique_files): planned = unique_files[version_id] source_key = f"{source_prefix}/{version_id}" destination_key = f"{destination_prefix}/{version_id}" destination_logical_key = f"{destination_logical_prefix}/{version_id}" planned_destination_key = str(planned.get("storage_key") or "").strip().lstrip("/") if planned_destination_key != destination_logical_key: raise DeployError( f"projected destination key drift for ContentVersion {version_id}" ) source_digest = download_s3_object_digest( bucket=source_bucket, key=source_key, region=source_region, ) expected_size = int(planned.get("size_bytes") or 0) actual_size = int(source_digest["size_bytes"]) if actual_size != expected_size: raise DeployError( f"snapshot blob size mismatch for ContentVersion {version_id}: " f"expected {expected_size}, got {actual_size}" ) source_sha = str(source_digest["sha256"]) destination = head_s3_object( bucket=destination_bucket, key=destination_key, storage_region=destination_region, ) destination_state = str(destination.get("state") or "") actual_destination_size = destination.get("actual_size_bytes") destination_size_bytes = ( -1 if actual_destination_size is None else int(actual_destination_size) ) if destination_state == "missing-s3-object": normalized_destination_state = "absent" elif destination_state == "present" and destination_size_bytes == actual_size: destination_sha = str(destination.get("metadata_sha256") or "").strip().lower() if not destination_sha: destination_digest = download_s3_object_digest( bucket=destination_bucket, key=destination_key, region=destination_region, ) destination_sha = str(destination_digest["sha256"]) if destination_sha != source_sha: raise DeployError( f"destination blob conflicts with ContentVersion {version_id}" ) normalized_destination_state = "present-matching" else: raise DeployError( f"destination blob conflicts with ContentVersion {version_id}: " f"{destination_state or 'unknown'}" ) result.append( { "source_content_version_id": version_id, "source_bucket": source_bucket, "source_key": source_key, "source_region": source_region, "destination_bucket": destination_bucket, "destination_key": destination_key, "destination_region": destination_region, "size_bytes": actual_size, "sha256": source_sha, "destination_state": normalized_destination_state, } ) return result def plan_blob_recoveries_or_block( planned_files: list[dict[str, Any]], *, source_storage: dict[str, Any], source_region: str, destination_storage: dict[str, Any], destination_region: str, ) -> tuple[list[dict[str, Any]], str]: try: return ( plan_blob_recoveries( planned_files, source_storage=source_storage, source_region=source_region, destination_storage=destination_storage, destination_region=destination_region, ), "", ) except DeployError as exc: return [], str(exc)[:1000] def bind_projection_ids( envv: dict[str, str], planned_files: list[dict[str, Any]], *, expected_rows: int, ) -> dict[str, Any]: raw_ids = [str(item.get("file_id") or "").strip() for item in planned_files] file_ids = sorted(file_id for file_id in raw_ids if file_id) blocker = "" if len(file_ids) != expected_rows: blocker = ( f"projection ID count {len(file_ids)} does not match " f"projectable row count {expected_rows}" ) elif len(set(file_ids)) != len(file_ids): blocker = "planned Salesforce projection IDs are not unique" if blocker or not file_ids: return {"file_ids": file_ids, "preimages": [], "blocker": blocker} preimages = load_file_rows( envv, limit=0, status="", file_ids=file_ids, ) returned_ids = sorted(str(row.get("Id") or "").strip() for row in preimages) if any(file_id not in set(file_ids) for file_id in returned_ids): raise DeployError("projection ID preflight returned an unexpected file ID") if preimages: blocker = "one or more deterministic Salesforce projection IDs already exist" return {"file_ids": file_ids, "preimages": preimages, "blocker": blocker} def copy_recovery_blob(blob: dict[str, Any]) -> None: destination_state = str(blob.get("destination_state") or "") if destination_state not in {"absent", "present-matching"}: raise DeployError("Salesforce recovery blob is not safe to apply") if destination_state == "absent": with tempfile.TemporaryDirectory(prefix="nexuscrm-salesforce-copy-") as tmp: source_path = Path(tmp) / "source" download_s3_object_to_path( bucket=str(blob["source_bucket"]), key=str(blob["source_key"]), region=str(blob["source_region"]), output_path=source_path, ) actual_size = source_path.stat().st_size actual_sha = file_sha256(source_path) if actual_size != int(blob["size_bytes"]) or actual_sha != str(blob["sha256"]): raise DeployError( "Salesforce snapshot blob changed after the bound dry-run" ) upload_s3_object( source_path=source_path, bucket=str(blob["destination_bucket"]), key=str(blob["destination_key"]), region=str(blob["destination_region"]), sha256=actual_sha, ) verify_s3_object_bytes( bucket=str(blob["destination_bucket"]), key=str(blob["destination_key"]), region=str(blob["destination_region"]), expected_size_bytes=int(blob["size_bytes"]), expected_sha256=str(blob["sha256"]), ) def projection_kwargs( manifest: dict[str, Any], *, snapshot_dir: Path, snapshot_manifest_name: str, ) -> dict[str, Any]: storage = manifest["storage"] return { "target_parent_keys": manifest["target_parent_keys"], "parent_object_api_name": str(manifest["parent_object_api_name"]), "target_table": str(manifest["target_table"]), "attachment_kind": str(manifest["attachment_kind"]), "source_type": "salesforce", "storage_provider": "s3", "storage_bucket": str(storage["bucket"]), "storage_key_prefix": str(storage["key_prefix"]), "logical_storage_key_prefix": str(storage["logical_key_prefix"]), "snapshot_dir": str(snapshot_dir), "snapshot_manifest_name": snapshot_manifest_name, "allow_non_test_env": True, "prefer_snapshot_storage": False, "create_only": True, "skip_if_targets_covered": False, } def load_target_manifest( target_manifest_path: Path, *, expected_target_manifest_sha256: str, ) -> tuple[dict[str, Any], str]: expected_sha = require_sha256( expected_target_manifest_sha256, "expected target manifest hash", ) manifest = load_hash_bound_json( target_manifest_path, expected_sha, "target manifest", ) require_exact_keys( manifest, { "schema_version", "operation", "environment", "approval_ref", "parent_object_api_name", "target_table", "attachment_kind", "snapshot_source_region", "target_parent_keys", "storage", }, "target manifest", ) if manifest["schema_version"] != 1: raise DeployError("target manifest schema_version must be 1") if manifest["operation"] != OPERATION: raise DeployError(f"target manifest operation must be {OPERATION}") for key in ( "environment", "approval_ref", "parent_object_api_name", "target_table", "attachment_kind", "snapshot_source_region", ): if not str(manifest[key] or "").strip(): raise DeployError(f"target manifest {key} is required") target_parent_keys = manifest["target_parent_keys"] if not isinstance(target_parent_keys, list) or not target_parent_keys: raise DeployError("target manifest target_parent_keys must be a non-empty array") normalized_target_keys = [str(value or "").strip() for value in target_parent_keys] if any(not value for value in normalized_target_keys): raise DeployError("target manifest target_parent_keys cannot contain empty values") if len(normalized_target_keys) != len(set(normalized_target_keys)): raise DeployError("target manifest target_parent_keys must be unique") manifest["target_parent_keys"] = normalized_target_keys storage = manifest["storage"] if not isinstance(storage, dict): raise DeployError("target manifest storage must be an object") require_exact_keys( storage, {"provider", "bucket", "region", "key_prefix"}, "target manifest storage", ) if str(storage["provider"] or "").strip().lower() != "s3": raise DeployError("target manifest storage provider must be s3") storage.update( { "provider": "s3", "bucket": str(storage["bucket"] or "").strip(), "region": str(storage["region"] or "").strip(), "key_prefix": str(storage["key_prefix"] or "").strip().strip("/"), } ) for key in ("bucket", "region", "key_prefix"): if not storage[key]: raise DeployError(f"target manifest storage {key} is required") return manifest, expected_sha def plan_salesforce_history_recovery( envv: dict[str, str], *, target_manifest_path: Path, expected_target_manifest_sha256: str, snapshot_dir: Path, snapshot_manifest_name: str, expected_snapshot_manifest_sha256: str, ) -> dict[str, Any]: manifest, target_manifest_sha = load_target_manifest( target_manifest_path, expected_target_manifest_sha256=expected_target_manifest_sha256, ) env_name = str(envv.get("ENV_NAME") or "").strip() if str(manifest["environment"]).strip() != env_name: raise DeployError("target manifest environment does not match runtime ENV_NAME") manifest["storage"] = bind_runtime_recovery_storage( envv, manifest["storage"], context="Salesforce recovery destination", ) snapshot_manifest_path = snapshot_dir / snapshot_manifest_name snapshot_manifest_sha = require_sha256( expected_snapshot_manifest_sha256, "expected snapshot manifest hash", ) snapshot_manifest = load_hash_bound_json( snapshot_manifest_path, snapshot_manifest_sha, "snapshot manifest", ) snapshot_objects = snapshot_manifest.get("objects") if not isinstance(snapshot_objects, list): raise DeployError( "Salesforce history recovery requires a direct snapshot manifest objects array" ) source_storage_authority = require_salesforce_history_archive_authority( snapshot_storage( manifest_objects=snapshot_objects, storage_provider="s3", fallback_storage_bucket="", fallback_storage_key_prefix="", ), source_region=str(manifest["snapshot_source_region"]), ) snapshot_files = bind_snapshot_input_files(snapshot_dir, snapshot_manifest) projection = repair_salesforce_file_projection_from_snapshot( envv, apply=False, **projection_kwargs( manifest, snapshot_dir=snapshot_dir, snapshot_manifest_name=snapshot_manifest_name, ), ) repair = projection.get("repair") or {} target_parent_ids = [ str(value or "").strip() for value in repair.get("target_parent_ids") or [] if str(value or "").strip() ] projectable_rows = int(repair.get("projectable_rows") or 0) planned_files = (repair.get("snapshot_result") or {}).get("planned_files") or [] projection_ids = bind_projection_ids( envv, planned_files, expected_rows=projectable_rows, ) projected_source_storage = repair.get("snapshot_source_storage") or {} if projected_source_storage: require_salesforce_history_archive_authority( projected_source_storage, source_region=str(manifest["snapshot_source_region"]), ) source_storage = source_storage_authority blobs: list[dict[str, Any]] = [] blob_blocker = "" if projection.get("status") == "would_repair" and planned_files: blobs, blob_blocker = plan_blob_recoveries_or_block( planned_files, source_storage=source_storage, source_region=str(manifest["snapshot_source_region"]), destination_storage=manifest["storage"], destination_region=str(manifest["storage"]["region"]), ) unique_version_ids = { str(planned.get("source_content_version_id") or "").strip() for planned in planned_files if str(planned.get("source_content_version_id") or "").strip() } ready = ( projection.get("status") == "would_repair" and projectable_rows > 0 and len(target_parent_ids) == len(manifest["target_parent_keys"]) and bool(unique_version_ids) and len(blobs) == len(unique_version_ids) and not blob_blocker and not projection_ids["blocker"] ) preimages: list[dict[str, Any]] = [] if target_parent_ids: preimages = load_file_rows( envv, limit=0, parent_object_api_name=str(manifest["parent_object_api_name"]), attachment_kind=str(manifest["attachment_kind"]), source_type="salesforce", status="", parent_ids=target_parent_ids, ) return { "schema_version": 1, "operation": OPERATION, "mode": "dry-run", "status": "ready" if ready else "blocked", "environment": env_name, "database": str(envv.get("TIDB_DATABASE") or ""), "approval_ref": str(manifest["approval_ref"]), "target_manifest_sha256": target_manifest_sha, "snapshot_manifest_sha256": snapshot_manifest_sha, "snapshot_manifest_name": snapshot_manifest_name, "snapshot_files": snapshot_files, "source_storage_authority": source_storage_authority, "target_parent_keys": manifest["target_parent_keys"], "target_parent_ids": target_parent_ids, "projectable_rows": projectable_rows, "projection_file_ids": projection_ids["file_ids"], "projection_id_preimages": projection_ids["preimages"], "projection_id_blocker": projection_ids["blocker"], "blobs": blobs, "blob_blocker": blob_blocker, "preimages": preimages, "projection": projection, } def require_applied_projection( projection: dict[str, Any], *, expected_rows: int, ) -> int: if ( projection.get("status") != "passed" or (projection.get("after") or {}).get("status") != "passed" ): raise DeployError("Salesforce history recovery final target audit did not pass") affected_rows = int((projection.get("repair") or {}).get("affected_rows") or 0) if affected_rows != expected_rows: raise DeployError( "Salesforce history recovery create-only projection inserted " f"{affected_rows} of {expected_rows} planned rows" ) return affected_rows def require_exact_projection_readback( rows: list[dict[str, Any]], *, planned_files: list[dict[str, Any]], storage: dict[str, Any], ) -> list[dict[str, Any]]: expected_by_id = { str(planned.get("file_id") or "").strip(): planned for planned in planned_files if str(planned.get("file_id") or "").strip() } actual_by_id = { str(row.get("Id") or "").strip(): row for row in rows if str(row.get("Id") or "").strip() } if len(expected_by_id) != len(planned_files): raise DeployError("Salesforce projection readback plan has missing or duplicate IDs") if set(actual_by_id) != set(expected_by_id) or len(actual_by_id) != len(rows): raise DeployError("Salesforce projection readback did not return the exact planned IDs") ordered_rows: list[dict[str, Any]] = [] for file_id in expected_by_id: planned = expected_by_id[file_id] row = actual_by_id[file_id] required_plan_fields = ( "name", "original_file_name", "parent_id", "parent_object_api_name", "attachment_kind", "source_type", "source_content_document_id", "source_content_version_id", "status", "storage_provider", "storage_bucket", "storage_key", "storage_path", ) missing_plan_fields = [ field for field in required_plan_fields if not str(planned.get(field) or "").strip() ] if missing_plan_fields: raise DeployError( f"Salesforce projection plan {file_id} is missing exact fields: " + ", ".join(missing_plan_fields) ) planned_size = parse_size_bytes(planned.get("size_bytes")) if planned_size is None: raise DeployError( f"Salesforce projection plan {file_id} has invalid size_bytes" ) expected_fields: dict[str, Any] = { "Name": str(planned.get("name") or ""), "OriginalFileName": str(planned.get("original_file_name") or ""), "ParentId": str(planned.get("parent_id") or ""), "ParentObjectApiName": str(planned.get("parent_object_api_name") or ""), "AttachmentKind": str(planned.get("attachment_kind") or ""), "MimeType": str(planned.get("mime_type") or ""), "FileExtension": str(planned.get("file_extension") or ""), "SourceType": str(planned.get("source_type") or ""), "SourceContentDocumentId": str( planned.get("source_content_document_id") or "" ), "SourceContentVersionId": str( planned.get("source_content_version_id") or "" ), "Status": str(planned.get("status") or ""), "StorageProvider": str(planned.get("storage_provider") or ""), "StorageBucket": str(planned.get("storage_bucket") or ""), "StorageKey": str(planned.get("storage_key") or ""), "StoragePath": str(planned.get("storage_path") or ""), "SizeBytes": planned_size, } if expected_fields["StorageProvider"] != str(storage.get("provider") or ""): raise DeployError( f"Salesforce projection plan {file_id} StorageProvider is outside recovery storage" ) if expected_fields["StorageBucket"] != str(storage.get("bucket") or ""): raise DeployError( f"Salesforce projection plan {file_id} StorageBucket is outside recovery storage" ) logical_prefix = str( storage.get("logical_key_prefix") or "" ).strip().strip("/") if not logical_prefix or not expected_fields["StorageKey"].startswith( logical_prefix + "/" ): raise DeployError( f"Salesforce projection plan {file_id} StorageKey is outside recovery storage" ) physical_prefix = str(storage.get("key_prefix") or "").strip().strip("/") expected_path_prefix = ( f"s3://{expected_fields['StorageBucket']}/{physical_prefix}/" ) if not physical_prefix or not expected_fields["StoragePath"].startswith( expected_path_prefix ): raise DeployError( f"Salesforce projection plan {file_id} StoragePath is outside recovery storage" ) for field, expected in expected_fields.items(): if field not in row: raise DeployError( f"Salesforce projection readback {file_id} is missing {field}" ) actual = ( parse_size_bytes(row[field]) if field == "SizeBytes" else str(row[field] or "") ) if actual != expected: raise DeployError( f"Salesforce projection readback {file_id} {field} mismatch" ) ordered_rows.append(row) return ordered_rows def apply_salesforce_history_recovery( envv: dict[str, str], *, target_manifest_path: Path, expected_target_manifest_sha256: str, snapshot_dir: Path, snapshot_manifest_name: str, expected_snapshot_manifest_sha256: str, dry_run_receipt_path: Path, expected_dry_run_receipt_sha256: str, backup_output: Path, apply_receipt_output: Path, actor: str, ) -> dict[str, Any]: normalized_actor = str(actor or "").strip() if not normalized_actor: raise DeployError("apply actor is required") if backup_output.exists() or apply_receipt_output.exists(): raise DeployError("backup and apply receipt outputs must not already exist") fresh_plan = plan_salesforce_history_recovery( envv, target_manifest_path=target_manifest_path, expected_target_manifest_sha256=expected_target_manifest_sha256, snapshot_dir=snapshot_dir, snapshot_manifest_name=snapshot_manifest_name, expected_snapshot_manifest_sha256=expected_snapshot_manifest_sha256, ) if fresh_plan["status"] != "ready": raise DeployError("Salesforce history recovery apply requires a ready plan") dry_run_sha = require_sha256( expected_dry_run_receipt_sha256, "expected dry-run receipt hash", ) bound_dry_run = load_hash_bound_json( dry_run_receipt_path, dry_run_sha, "dry-run receipt", ) if bound_dry_run != fresh_plan: raise DeployError("live recovery plan no longer matches the bound dry-run receipt") manifest, target_manifest_sha = load_target_manifest( target_manifest_path, expected_target_manifest_sha256=expected_target_manifest_sha256, ) manifest["storage"] = bind_runtime_recovery_storage( envv, manifest["storage"], context="Salesforce recovery destination", ) backup_payload = { "schema_version": 1, "operation": OPERATION, "environment": fresh_plan["environment"], "target_manifest_sha256": target_manifest_sha, "snapshot_manifest_sha256": fresh_plan["snapshot_manifest_sha256"], "dry_run_receipt_sha256": dry_run_sha, "source_storage_authority": fresh_plan["source_storage_authority"], "target_parent_keys": fresh_plan["target_parent_keys"], "target_parent_ids": fresh_plan["target_parent_ids"], "preimages": fresh_plan["preimages"], "blobs": fresh_plan["blobs"], "snapshot_files": fresh_plan["snapshot_files"], "projection_file_ids": fresh_plan["projection_file_ids"], "projection_id_preimages": fresh_plan["projection_id_preimages"], } backup_sha = write_json_exclusive(backup_output, backup_payload) with tempfile.TemporaryDirectory(prefix="nexuscrm-bound-snapshot-") as tmp: bound_snapshot_dir = Path(tmp) / "snapshot" materialize_bound_snapshot( source_dir=snapshot_dir, destination_dir=bound_snapshot_dir, manifest_name=snapshot_manifest_name, expected_manifest_sha256=fresh_plan["snapshot_manifest_sha256"], snapshot_files=fresh_plan["snapshot_files"], ) projection_preflight = bind_projection_ids( envv, [ {"file_id": file_id} for file_id in fresh_plan["projection_file_ids"] ], expected_rows=int(fresh_plan["projectable_rows"]), ) if projection_preflight != { "file_ids": fresh_plan["projection_file_ids"], "preimages": [], "blocker": "", }: raise DeployError( "Salesforce projection IDs changed after the bound dry-run" ) def copy_bound_blobs_before_commit() -> None: for blob in fresh_plan["blobs"]: copy_recovery_blob(blob) applied_projection = repair_salesforce_file_projection_from_snapshot( envv, apply=True, expected_parent_ids=fresh_plan["target_parent_ids"], before_projection_commit=copy_bound_blobs_before_commit, **projection_kwargs( manifest, snapshot_dir=bound_snapshot_dir, snapshot_manifest_name=snapshot_manifest_name, ), ) affected_rows = require_applied_projection( applied_projection, expected_rows=int(fresh_plan["projectable_rows"]), ) planned_files = ( (((fresh_plan.get("projection") or {}).get("repair") or {}).get("snapshot_result") or {}) .get("planned_files") or [] ) projection_readback = require_exact_projection_readback( load_file_rows( envv, limit=0, status="", file_ids=fresh_plan["projection_file_ids"], ), planned_files=planned_files, storage=manifest["storage"], ) receipt = { "schema_version": 1, "operation": OPERATION, "status": "applied_pending_authenticated_readback", "environment": fresh_plan["environment"], "actor": normalized_actor, "approval_ref": fresh_plan["approval_ref"], "target_manifest_sha256": target_manifest_sha, "snapshot_manifest_sha256": fresh_plan["snapshot_manifest_sha256"], "dry_run_receipt_sha256": dry_run_sha, "backup_sha256": backup_sha, "source_storage_authority": fresh_plan["source_storage_authority"], "target_parent_keys": fresh_plan["target_parent_keys"], "target_parent_ids": fresh_plan["target_parent_ids"], "projectable_rows": fresh_plan["projectable_rows"], "affected_rows": affected_rows, "projection_readback": projection_readback, "blob_count": len(fresh_plan["blobs"]), "blobs": fresh_plan["blobs"], "projection": applied_projection, "post_readback_required": "authenticated-backend-download", } apply_receipt_sha = write_json_exclusive(apply_receipt_output, receipt) return {**receipt, "apply_receipt_sha256": apply_receipt_sha} def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Plan or apply an exact, hash-bound historical Salesforce Deal " "document projection recovery." ) ) parser.add_argument("--env-file", required=True) parser.add_argument("--target-manifest", required=True) parser.add_argument("--expected-target-manifest-sha256", required=True) parser.add_argument("--snapshot-dir", required=True) parser.add_argument("--snapshot-manifest-name", default="manifest.json") parser.add_argument("--expected-snapshot-manifest-sha256", required=True) parser.add_argument("--receipt-output", required=True) parser.add_argument("--mode", choices=("dry-run", "apply"), default="dry-run") parser.add_argument("--dry-run-receipt", default="") parser.add_argument("--expected-dry-run-receipt-sha256", default="") parser.add_argument("--backup-output", default="") parser.add_argument("--actor", default="") return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) envv = load_env_file(Path(args.env_file).expanduser().resolve()) target_manifest_path = Path(args.target_manifest).expanduser().resolve() snapshot_dir = Path(args.snapshot_dir).expanduser().resolve() receipt_output = Path(args.receipt_output).expanduser().resolve() common_kwargs = { "target_manifest_path": target_manifest_path, "expected_target_manifest_sha256": args.expected_target_manifest_sha256, "snapshot_dir": snapshot_dir, "snapshot_manifest_name": args.snapshot_manifest_name, "expected_snapshot_manifest_sha256": args.expected_snapshot_manifest_sha256, } if args.mode == "apply": missing = [ name for name, value in ( ("--dry-run-receipt", args.dry_run_receipt), ( "--expected-dry-run-receipt-sha256", args.expected_dry_run_receipt_sha256, ), ("--backup-output", args.backup_output), ("--actor", args.actor), ) if not str(value or "").strip() ] if missing: raise DeployError(f"apply mode requires: {', '.join(missing)}") summary = apply_salesforce_history_recovery( envv, **common_kwargs, dry_run_receipt_path=Path(args.dry_run_receipt).expanduser().resolve(), expected_dry_run_receipt_sha256=args.expected_dry_run_receipt_sha256, backup_output=Path(args.backup_output).expanduser().resolve(), apply_receipt_output=receipt_output, actor=args.actor, ) print(json.dumps(summary, indent=2, sort_keys=True)) return 0 summary = plan_salesforce_history_recovery(envv, **common_kwargs) write_json_exclusive(receipt_output, summary) print(json.dumps(summary, indent=2, sort_keys=True)) return 0 if summary["status"] == "ready" else 1 if __name__ == "__main__": raise SystemExit(main())