#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re from pathlib import Path from typing import Any from common import DeployError, db_query_scalar, load_env_file DEFAULT_PARENT_OBJECT_API_NAME = "Order" DEFAULT_TARGET_TABLE = "order" DEFAULT_SYSTEM_FILE_TABLE = "_System_File" DEFAULT_ATTACHMENT_KIND = "archived_order" DEFAULT_SOURCE_TYPE = "salesforce" DEFAULT_STORAGE_PROVIDER = "s3" DEFAULT_STORAGE_BUCKET = "nexuscrm-sf-export-604175033442-us-west-2" DEFAULT_STORAGE_KEY_PREFIX = "salesforce/content/blobs" DEFAULT_STATUS = "Active" DEFAULT_TARGET_PARENT_KEY_COLUMNS = ( "Id", "Name", "OrderNumber", "Order_Number__c", "ContractNumber", ) DEFAULT_FILE_LIMIT = 20 SAFE_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") TEST_SLOT_ENV_NAMES = {"test-test1", "test-test2", "test-test3"} PROBLEM_STATES = { "ambiguous-target", "missing-target", "missing-file", "missing-source-content-version-id", "missing-storage-location", "unexpected-storage-provider", "unexpected-storage-bucket", "unexpected-storage-key", "unexpected-storage-path", } def sql_string_literal(value: str) -> str: return "'" + str(value).replace("'", "''") + "'" def sanitize_identifier(value: str) -> str: value = str(value or "").strip() if not SAFE_IDENTIFIER_RE.match(value): raise DeployError(f"unsafe SQL identifier: {value!r}") return value def split_csv_list(raw: str) -> list[str]: return [item.strip() for item in str(raw or "").split(",") if item.strip()] def unique_ordered(values: list[str]) -> list[str]: result: list[str] = [] seen: set[str] = set() for raw in values: value = str(raw or "").strip() if not value or value in seen: continue result.append(value) seen.add(value) return result def is_test_slot_env_name(env_name: str) -> bool: return str(env_name or "").strip().lower() in TEST_SLOT_ENV_NAMES def require_test_slot_env(envv: dict[str, str], *, allow_non_test_env: bool) -> None: if allow_non_test_env: return env_name = str(envv.get("ENV_NAME") or "") if not is_test_slot_env_name(env_name): raise DeployError( "Salesforce file projection audit is test-slot only by default; " f"got ENV_NAME={env_name!r}" ) def normalize_storage_key_prefix(value: str) -> str: return str(value or "").strip().strip("/") def normalize_logical_storage_key_prefix(value: str) -> str: prefix = str(value or "").strip() if not prefix: return "" if prefix.startswith("/") or prefix.endswith("/") or "\\" in prefix: raise DeployError("logical storage key prefix is not canonical") if any(part in {"", ".", ".."} for part in prefix.split("/")): raise DeployError("logical storage key prefix is not canonical") return prefix def column_lookup(columns: list[str]) -> dict[str, str]: return {column.lower(): column for column in columns} def query_lines(envv: dict[str, str], sql: str) -> list[str]: raw = db_query_scalar(envv, sql, database=envv["TIDB_DATABASE"]) return [line.strip() for line in raw.splitlines() if line.strip()] def query_json_rows(envv: dict[str, str], sql: str) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for line in query_lines(envv, sql): if line.startswith("{"): payload = json.loads(line) else: try: payload = json.loads(bytes.fromhex(line).decode("utf-8")) except ValueError as exc: raise DeployError(f"invalid hex JSON row from TiDB: {line[:80]!r}") from exc if isinstance(payload, dict): rows.append(payload) return rows def fetch_table_columns(envv: dict[str, str], table_name: str) -> list[str]: sql = f""" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = {sql_string_literal(table_name)} ORDER BY ORDINAL_POSITION """.strip() return query_lines(envv, sql) def json_object_expr(alias: str, columns: list[str]) -> str: lookup = column_lookup(columns) entries: list[str] = [] for requested in ( "Id", "Name", "OrderNumber", "Order_Number__c", "ContractNumber", "OriginalFileName", "ParentId", "ParentObjectApiName", "AttachmentKind", "SourceType", "SourceContentVersionId", "Status", "StorageProvider", "StorageBucket", "StorageKey", "StoragePath", "SizeBytes", "CreatedDate", "LastModifiedDate", ): actual = lookup.get(requested.lower()) if actual: entries.append( f"{sql_string_literal(requested)}, CAST({alias}.`{sanitize_identifier(actual)}` AS CHAR)" ) if not entries: raise DeployError("cannot build JSON_OBJECT without projected columns") return "HEX(CAST(JSON_OBJECT(" + ", ".join(entries) + ") AS CHAR))" def available_target_key_columns(target_columns: list[str]) -> list[str]: lookup = column_lookup(target_columns) return [ lookup[column.lower()] for column in DEFAULT_TARGET_PARENT_KEY_COLUMNS if column.lower() in lookup ] def build_target_query(*, target_table: str, target_columns: list[str], target_key: str) -> str: key_columns = available_target_key_columns(target_columns) if not key_columns: raise DeployError( f"target table {target_table!r} lacks all key columns: " f"{', '.join(DEFAULT_TARGET_PARENT_KEY_COLUMNS)}" ) predicates = [ f"CAST(parent_record.`{sanitize_identifier(column)}` AS CHAR) = {sql_string_literal(target_key)}" for column in key_columns ] return f""" SELECT {json_object_expr("parent_record", target_columns)} FROM `{sanitize_identifier(target_table)}` parent_record WHERE {" OR ".join(predicates)} LIMIT 2 """.strip() def build_system_file_query( *, system_file_table: str, system_file_columns: list[str], parent_id: str, parent_object_api_name: str, attachment_kind: str, source_type: str, status: str, storage_provider: str, storage_bucket: str, limit: int, ) -> str: lookup = column_lookup(system_file_columns) required = [ "ParentId", "ParentObjectApiName", "AttachmentKind", "SourceType", "SourceContentVersionId", "Status", ] missing = [column for column in required if column.lower() not in lookup] if storage_provider and "storageprovider" not in lookup: missing.append("StorageProvider") if storage_bucket and "storagebucket" not in lookup: missing.append("StorageBucket") if "storagekey" not in lookup and "storagepath" not in lookup: missing.append("StorageKey/StoragePath") if missing: raise DeployError( f"system file table {system_file_table!r} missing required columns: " f"{', '.join(missing)}" ) filters = [ f"sf.`{sanitize_identifier(lookup['parentid'])}` = {sql_string_literal(parent_id)}", ( f"sf.`{sanitize_identifier(lookup['parentobjectapiname'])}` = " f"{sql_string_literal(parent_object_api_name)}" ), f"sf.`{sanitize_identifier(lookup['attachmentkind'])}` = {sql_string_literal(attachment_kind)}", f"sf.`{sanitize_identifier(lookup['sourcetype'])}` = {sql_string_literal(source_type)}", f"sf.`{sanitize_identifier(lookup['status'])}` = {sql_string_literal(status)}", f"sf.`{sanitize_identifier(lookup['sourcecontentversionid'])}` IS NOT NULL", f"TRIM(CAST(sf.`{sanitize_identifier(lookup['sourcecontentversionid'])}` AS CHAR)) <> ''", ] if storage_provider: filters.append( f"sf.`{sanitize_identifier(lookup['storageprovider'])}` = {sql_string_literal(storage_provider)}" ) if storage_bucket: filters.append( f"sf.`{sanitize_identifier(lookup['storagebucket'])}` = {sql_string_literal(storage_bucket)}" ) order_by = "" if "createddate" in lookup: order_by = f"\nORDER BY sf.`{sanitize_identifier(lookup['createddate'])}` DESC" limit = max(1, min(int(limit), 100)) return f""" SELECT {json_object_expr("sf", system_file_columns)} FROM `{sanitize_identifier(system_file_table)}` sf WHERE {" AND ".join(filters)}{order_by} LIMIT {limit} """.strip() def storage_location(row: dict[str, Any]) -> str: return str(row.get("StorageKey") or row.get("StoragePath") or "").strip() def check_file_metadata( row: dict[str, Any], *, storage_provider: str, storage_bucket: str, storage_key_prefix: str, logical_storage_key_prefix: str, ) -> str: source_content_version_id = str(row.get("SourceContentVersionId") or "").strip() if not source_content_version_id: return "missing-source-content-version-id" provider = str(row.get("StorageProvider") or "").strip() if storage_provider and provider != storage_provider: return "unexpected-storage-provider" bucket = str(row.get("StorageBucket") or "").strip() if storage_bucket and bucket != storage_bucket: return "unexpected-storage-bucket" storage_key = str(row.get("StorageKey") or "").strip() storage_path = str(row.get("StoragePath") or "").strip() if not storage_key and not storage_path: return "missing-storage-location" logical_prefix = normalize_logical_storage_key_prefix( logical_storage_key_prefix ) expected_storage_key = ( f"{logical_prefix}/{source_content_version_id}" if logical_prefix else source_content_version_id ) if storage_key != expected_storage_key: return "unexpected-storage-key" physical_prefix = normalize_storage_key_prefix(storage_key_prefix) physical_key = ( f"{physical_prefix}/{source_content_version_id}" if physical_prefix else source_content_version_id ) expected_storage_path = f"s3://{storage_bucket}/{physical_key}" if storage_path != expected_storage_path: return "unexpected-storage-path" return "covered" def target_match_columns(target_row: dict[str, Any], target_key: str) -> list[str]: return [ column for column in DEFAULT_TARGET_PARENT_KEY_COLUMNS if str(target_row.get(column) or "").strip() == target_key ] def audit_target_key( envv: dict[str, str], *, target_key: str, target_table: str, target_columns: list[str], system_file_table: str, system_file_columns: list[str], parent_object_api_name: str, attachment_kind: str, source_type: str, status: str, storage_provider: str, storage_bucket: str, storage_key_prefix: str, logical_storage_key_prefix: str, file_limit: int, ) -> dict[str, Any]: target_rows = query_json_rows( envv, build_target_query( target_table=target_table, target_columns=target_columns, target_key=target_key, ), ) if not target_rows: return { "target_key": target_key, "state": "missing-target", "target_found": False, "file_count": 0, "covered_file_count": 0, "files": [], } if len(target_rows) != 1: return { "target_key": target_key, "state": "ambiguous-target", "target_found": False, "matched_parent_ids": [ str(row.get("Id") or "").strip() for row in target_rows ], "file_count": 0, "covered_file_count": 0, "files": [], } target_row = target_rows[0] parent_id = str(target_row.get("Id") or "").strip() file_rows = query_json_rows( envv, build_system_file_query( system_file_table=system_file_table, system_file_columns=system_file_columns, parent_id=parent_id, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, source_type=source_type, status=status, storage_provider=storage_provider, storage_bucket=storage_bucket, limit=file_limit, ), ) checked_files = [] covered_count = 0 first_problem = "" for row in file_rows: state = check_file_metadata( row, storage_provider=storage_provider, storage_bucket=storage_bucket, storage_key_prefix=storage_key_prefix, logical_storage_key_prefix=logical_storage_key_prefix, ) if state == "covered": covered_count += 1 elif not first_problem: first_problem = state checked_files.append( { "id": str(row.get("Id") or ""), "name": str(row.get("Name") or row.get("OriginalFileName") or ""), "source_content_version_id": str(row.get("SourceContentVersionId") or ""), "storage_provider": str(row.get("StorageProvider") or ""), "storage_bucket": str(row.get("StorageBucket") or ""), "storage_location": storage_location(row), "state": state, } ) target_state = "covered" if covered_count > 0 else first_problem or "missing-file" return { "target_key": target_key, "state": target_state, "target_found": True, "parent_id": parent_id, "matched_columns": target_match_columns(target_row, target_key), "file_count": len(file_rows), "covered_file_count": covered_count, "files": checked_files, } def audit_salesforce_file_projection( envv: dict[str, str], *, target_parent_keys: list[str], parent_object_api_name: str = DEFAULT_PARENT_OBJECT_API_NAME, target_table: str = DEFAULT_TARGET_TABLE, system_file_table: str = DEFAULT_SYSTEM_FILE_TABLE, attachment_kind: str = DEFAULT_ATTACHMENT_KIND, source_type: str = DEFAULT_SOURCE_TYPE, storage_provider: str = DEFAULT_STORAGE_PROVIDER, storage_bucket: str = DEFAULT_STORAGE_BUCKET, storage_key_prefix: str = DEFAULT_STORAGE_KEY_PREFIX, logical_storage_key_prefix: str = "", status: str = DEFAULT_STATUS, file_limit: int = DEFAULT_FILE_LIMIT, allow_non_test_env: bool = False, ) -> dict[str, Any]: require_test_slot_env(envv, allow_non_test_env=allow_non_test_env) target_keys = unique_ordered(target_parent_keys) if not target_keys: raise DeployError("at least one --target-parent-key is required") target_columns = fetch_table_columns(envv, target_table) system_file_columns = fetch_table_columns(envv, system_file_table) targets = [ audit_target_key( envv, target_key=target_key, target_table=target_table, target_columns=target_columns, system_file_table=system_file_table, system_file_columns=system_file_columns, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, source_type=source_type, status=status, storage_provider=storage_provider, storage_bucket=storage_bucket, storage_key_prefix=storage_key_prefix, logical_storage_key_prefix=logical_storage_key_prefix, file_limit=file_limit, ) for target_key in target_keys ] problems = [target for target in targets if str(target.get("state")) in PROBLEM_STATES] covered = [target for target in targets if str(target.get("state")) == "covered"] return { "status": "passed" if not problems else "failed", "mode": "audit-only", "env_name": envv.get("ENV_NAME", ""), "public_base_url": envv.get("PUBLIC_BASE_URL", ""), "filters": { "parent_object_api_name": parent_object_api_name, "target_table": target_table, "system_file_table": system_file_table, "attachment_kind": attachment_kind, "source_type": source_type, "status": status, "storage_provider": storage_provider, "storage_bucket": storage_bucket, "storage_key_prefix": normalize_storage_key_prefix(storage_key_prefix), "logical_storage_key_prefix": normalize_logical_storage_key_prefix( logical_storage_key_prefix ), }, "target_key_count": len(target_keys), "target_keys_with_files": len(covered), "target_keys_missing_files": len(target_keys) - len(covered), "problem_count": len(problems), "problem_samples": problems[:10], "targets": targets, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Audit Salesforce migrated file projection coverage in _System_File." ) parser.add_argument("--env-file", required=True) parser.add_argument("--parent-object-api-name", default=DEFAULT_PARENT_OBJECT_API_NAME) parser.add_argument("--target-table", default=DEFAULT_TARGET_TABLE) parser.add_argument("--system-file-table", default=DEFAULT_SYSTEM_FILE_TABLE) parser.add_argument("--attachment-kind", default=DEFAULT_ATTACHMENT_KIND) parser.add_argument("--source-type", default=DEFAULT_SOURCE_TYPE) parser.add_argument("--storage-provider", default=DEFAULT_STORAGE_PROVIDER) parser.add_argument("--storage-bucket", default=DEFAULT_STORAGE_BUCKET) parser.add_argument("--storage-key-prefix", default=DEFAULT_STORAGE_KEY_PREFIX) parser.add_argument("--logical-storage-key-prefix", default="") parser.add_argument("--status", default=DEFAULT_STATUS) parser.add_argument("--file-limit", type=int, default=DEFAULT_FILE_LIMIT) parser.add_argument( "--target-parent-key", action="append", default=[], help="Parent record key to audit. May be repeated or comma-separated.", ) parser.add_argument("--allow-non-test-env", action="store_true") parser.add_argument("--summary-json") parser.add_argument("--fail-on-problems", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() envv = load_env_file(Path(args.env_file).expanduser().resolve()) summary = audit_salesforce_file_projection( envv, target_parent_keys=[ item for raw in args.target_parent_key for item in split_csv_list(raw) ], parent_object_api_name=args.parent_object_api_name, target_table=args.target_table, system_file_table=args.system_file_table, attachment_kind=args.attachment_kind, source_type=args.source_type, storage_provider=args.storage_provider, storage_bucket=args.storage_bucket, storage_key_prefix=args.storage_key_prefix, logical_storage_key_prefix=args.logical_storage_key_prefix, status=args.status, file_limit=args.file_limit, allow_non_test_env=args.allow_non_test_env, ) output = json.dumps(summary, indent=2, ensure_ascii=False) if args.summary_json: Path(args.summary_json).write_text(output + "\n", encoding="utf-8") print(output) if args.fail_on_problems and summary.get("status") != "passed": return 1 return 0 if __name__ == "__main__": raise SystemExit(main())