#!/usr/bin/env python3 """Bind the ``Account_Notify_Approver_On_Assignment`` Flow (L2 pack) to the StepApproval event of each regional Account approval process, as a PROCESS-LEVEL ``ProcessAction`` (``ProcessNodeId`` NULL). Why process-level: the approval engine injects ``Approval.ApproverUserIds`` with the POST-GATING next-step approver ids only on the process-level StepApproval dispatch (step advance). A node-scoped binding fires with ``nil`` actors and resolves to an empty recipient set, so it would email nobody. Process-level also means a conditionally-skipped step (e.g. Legal when ``Vertical__c`` is not ``Crypto Exchange`` / ``Gambling & Gaming``) never advances and never notifies. Unlike ``apply_credits_approval_seed.py`` this applier does NOT define or modify any ``ProcessDefinition`` / ``ProcessNode``: the regional Account approval processes are built in the Setup UI. It only ADDS the notify action, matched to the EXISTING **active** process by ``DeveloperName`` + ``TableEnumOrId='Account'`` (``State='Active' AND IsActive=1``). FAIL-CLOSED by default: any region left unbound (absent / inactive / wrong-object process, e.g. after a blue-green reset) rolls back and exits 1, so a partial or zero binding can never report success. Pass ``--allow-partial-bindings`` only where a partial binding is intentionally acceptable. Idempotent: an existing action for the same (process, event, flow) is updated in place; a fresh one gets a deterministic id. Prod note: there is NO automated or Setup-UI path for this binding on prod yet. Multi-prod applies only the installed pack + reference-data (not this seed), and the Setup UI creates only NODE-SCOPED StepApproval actions -- it CANNOT create the process-level (ProcessNodeId-NULL) binding this Flow requires. Prod binding therefore requires running this applier directly against the prod DB as a controlled operator step (or a future automated prod-wiring path); the same prod-delivery gap applies to the Credits chains. This seed makes the binding durable on staging (EXP applies neither the reference-data template nor these UI-built processes). """ from __future__ import annotations import argparse import hashlib import json import os import ssl from pathlib import Path from typing import Any def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Bind the Account notify Flow to the regional Account approval processes." ) parser.add_argument("--seed-file", required=True, type=Path) parser.add_argument("--validate-only", action="store_true") parser.add_argument( "--allow-partial-bindings", action="store_true", help=( "Opt OUT of fail-closed. DEFAULT is fail-closed: any declared region left unbound " "(absent / inactive / wrong-object process, e.g. after a blue-green reset) rolls " "back and exits 1, because the binding is the only checked-in proof the feature is " "live and prod binding is a manual operator step. Pass this ONLY for environments " "where a partial binding is intentionally acceptable." ), ) return parser.parse_args() def required_env(name: str) -> str: value = os.environ.get(name) if not value: raise SystemExit(f"missing required environment variable: {name}") return value def tls_enabled(value: str) -> bool: return str(value).strip().lower() not in ("0", "false", "no", "off", "") def connect(): import pymysql kwargs: dict[str, Any] = { "host": required_env("TIDB_HOST"), "port": int(required_env("TIDB_PORT")), "user": required_env("TIDB_USER"), "password": os.environ.get("TIDB_PASSWORD", ""), "database": required_env("TIDB_DATABASE"), "charset": "utf8mb4", "autocommit": False, "connect_timeout": 10, "read_timeout": 60, "write_timeout": 60, "cursorclass": pymysql.cursors.DictCursor, } if tls_enabled(os.environ.get("TIDB_TLS", "true")): kwargs["ssl"] = ssl.create_default_context() return pymysql.connect(**kwargs) def stable_action_id(process_dev: str, event: str, api_name: str) -> str: digest = hashlib.sha1(f"{process_dev}:{event}:{api_name}".encode("utf-8")).hexdigest()[:24] return f"procact_account_notify_{digest}" def validate(seed: dict[str, Any]) -> list[str]: errors: list[str] = [] actions = seed.get("processActions") or [] if not actions: errors.append("processActions is empty") for field in ("targetObject", "emailTemplateDeveloperName", "emailTemplateId"): if not str(seed.get(field) or "").strip(): errors.append(f"{field} is required") seen: set[tuple[str, str, str]] = set() for i, action in enumerate(actions): dev = str(action.get("processDeveloperName") or "").strip() event = str(action.get("actionEvent") or "").strip() api = str(action.get("apiName") or "").strip() if not dev: errors.append(f"processActions[{i}] missing processDeveloperName") if str(action.get("actionType") or "").strip() != "Flow": errors.append(f"processActions[{i}] actionType must be 'Flow'") if not api: errors.append(f"processActions[{i}] missing apiName (target Flow)") # process-level only: a bound ProcessNodeId would resolve to no recipients if str(action.get("processNodeId") or "").strip(): errors.append(f"processActions[{i}] must be process-level (no processNodeId)") keytuple = (dev, event, api) if keytuple in seen: errors.append(f"processActions[{i}] duplicate binding {keytuple}") seen.add(keytuple) return errors def existing_process_id(cur, developer_name: str, target_object: str) -> str | None: # Approval-process identity is OBJECT-SCOPED (DeveloperName + TableEnumOrId): a # same-name process on another object must never receive this Account Flow action # (that would leave the Account process unbound while the gate still "passed"). # Require BOTH State='Active' AND IsActive=1: a blue-green reset can leave the # process present-but-inactive, and a disabled process is not a valid target. cur.execute( "SELECT Id FROM ProcessDefinition " "WHERE DeveloperName = %s AND TableEnumOrId = %s " " AND State = 'Active' AND IsActive = 1 " "LIMIT 1 FOR UPDATE", (developer_name, target_object), ) row = cur.fetchone() return str(row["Id"]) if row else None def check_template(cur, developer_name: str, expected_id: str) -> str: # The engine resolves sendEmail templates by ID only (GetRaw), so if reference-data # reconciliation left this template under a DIFFERENT id, the Flow's hardcoded # templateId never resolves and the notification fails silently at send time. # Surface that as a loud deploy failure instead. Returns: ok|wrong-id|inactive|absent. cur.execute( "SELECT Id, IsActive FROM _System_EmailTemplate " "WHERE DeveloperName = %s AND (IsDeleted = 0 OR IsDeleted IS NULL) LIMIT 1", (developer_name,), ) row = cur.fetchone() if not row: return "absent" if str(row["Id"]) != expected_id: return "wrong-id" active = row.get("IsActive") if active in (0, None, False) or str(active).strip().lower() in ("0", "false", ""): return "inactive" return "ok" def existing_action_id(cur, process_id: str, event: str, api_name: str) -> str | None: cur.execute( """ SELECT Id FROM ProcessAction WHERE ProcessDefinitionId = %s AND ActionEvent = %s AND ApiName = %s AND (IsDeleted = 0 OR IsDeleted IS NULL) LIMIT 1 """, (process_id, event, api_name), ) row = cur.fetchone() return str(row["Id"]) if row else None def upsert_action(cur, process_id: str, action: dict[str, Any]) -> None: dev = action["processDeveloperName"] event = str(action.get("actionEvent") or "").strip() api = str(action.get("apiName") or "").strip() action_id = existing_action_id(cur, process_id, event, api) or stable_action_id(dev, event, api) cur.execute( """ INSERT INTO ProcessAction (Id, Name, ProcessDefinitionId, ProcessNodeId, ActionType, ActionEvent, FieldApiName, FieldValue, EmailTemplateId, ApiName, `Order`, IsDeleted, CreatedDate, LastModifiedDate) VALUES (%s, %s, %s, NULL, 'Flow', %s, '', '', '', %s, %s, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON DUPLICATE KEY UPDATE Name=VALUES(Name), ProcessDefinitionId=VALUES(ProcessDefinitionId), ProcessNodeId=NULL, ActionType='Flow', ActionEvent=VALUES(ActionEvent), ApiName=VALUES(ApiName), `Order`=VALUES(`Order`), IsDeleted=0, LastModifiedDate=CURRENT_TIMESTAMP """, ( action_id, str(action.get("name") or f"{dev} - Notify Next Approver"), process_id, event, api, int(action.get("order") or 0), ), ) def main() -> int: args = parse_args() seed = json.loads(args.seed_file.read_text(encoding="utf-8")) errors = validate(seed) if errors: for err in errors: print(f"ERROR: {err}") return 1 if args.validate_only: print(f"account notify binding seed OK ({len(seed.get('processActions') or [])} bindings)") return 0 target_object = str(seed["targetObject"]).strip() template_dev = str(seed["emailTemplateDeveloperName"]).strip() expected_template_id = str(seed["emailTemplateId"]).strip() conn = connect() applied: list[str] = [] skipped: list[str] = [] template_status = "no-bindings" try: with conn.cursor() as cur: for action in seed["processActions"]: dev = str(action["processDeveloperName"]).strip() process_id = existing_process_id(cur, dev, target_object) if not process_id: skipped.append(dev) continue upsert_action(cur, process_id, action) applied.append(dev) if applied: template_status = check_template(cur, template_dev, expected_template_id) print( "account notify binding: " f"bound={len(applied)} {applied} | skipped={len(skipped)} {skipped} | " f"template[{template_dev}]={template_status}" ) if skipped: print( f"WARNING: no active object-scoped {target_object} ProcessDefinition for: " + ", ".join(skipped) + " -- these regions have NO notify binding." ) # Decide BEFORE committing so a failing deploy never persists partial or invalid # ProcessAction rows -- evaluate every fail condition, then roll back if any holds. failures: list[str] = [] # An absent / wrong-id / inactive template means the Flow's hardcoded templateId # cannot resolve, so a committed binding would silently disable notifications. The # staging hook runs this only after reference-data (which owns the template) has been # applied, so an absent template here is a real failure, not a benign ordering gap. if template_status in ("absent", "wrong-id", "inactive"): failures.append( f"email template {template_dev!r} ({expected_template_id}) is {template_status}; " "the Flow's hardcoded templateId will not resolve to an active template, so " "notifications would fail silently at send time" ) # Staging is the only checked-in proof the binding exists: any unbound declared region # (absent / inactive / wrong-object) must fail, not just an all-zero run. if skipped and not args.allow_partial_bindings: failures.append( f"{len(skipped)} declared region(s) had no active object-scoped " f"{target_object} approval process and were left unbound: {skipped}; every " "declared region must bind (pass --allow-partial-bindings to opt out of " "fail-closed)" ) if failures: conn.rollback() for reason in failures: print(f"ERROR: {reason}. Rolled back -- no ProcessAction rows persisted. Failing closed.") return 1 conn.commit() finally: conn.close() return 0 if __name__ == "__main__": raise SystemExit(main())