#!/usr/bin/env python3 import json import os import re import ssl import time import urllib.parse import uuid from datetime import date, datetime, timedelta, timezone from pathlib import Path import boto3 import pymysql RULE_VERSION = "active_account_then_open_opportunity_v2" STATE_ID = 1 TENANT_ID_PATTERN = re.compile(r"^[0-9]+$") SYNC_ID_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{1,96}$") ACTIVE_TENANT_IDS_SQL = """ SELECT id FROM tenants WHERE status = 'active' AND kind = 'external' AND sales_force_region = 'APAC' AND country NOT IN ('JP') AND plan NOT IN ('free_trial', 'new_user', 'closing') """ CST_CURRENT_DAY_START_UTC_SQL = ( "DATE_SUB(DATE(DATE_ADD(UTC_TIMESTAMP(), INTERVAL 8 HOUR)), INTERVAL 8 HOUR)" ) NEW_SIGNUP_TENANT_IDS_SQL = f""" SELECT id FROM tenants WHERE created_at >= DATE_SUB({CST_CURRENT_DAY_START_UTC_SQL}, INTERVAL 1 DAY) AND created_at < {CST_CURRENT_DAY_START_UTC_SQL} AND country_region = 'APAC' AND country NOT IN ('JP') """ SNAPSHOT_COLUMNS = ( "sync_id", "tenant_id", "org_matched", "account_id", "account_matched", "account_approved", "account_owner_id", "account_owner_active", "lead_id", "lead_owner_id", "lead_owner_active", "selected_opportunity_id", "opportunity_stage", "opportunity_has_open_activity", "opportunity_owner_id", "opportunity_owner_active", "effective_owner_id", "effective_owner_email", "effective_owner_source", "effective_owner_active", "decision_status", "rule_version", "source_read_at", "snapshot_synced_at", ) _secrets_client = None def env(name, default=None, required=False): value = os.environ.get(name, default) if required and (value is None or str(value).strip() == ""): raise RuntimeError(f"{name} is required") return value def json_default(value): if isinstance(value, datetime): if value.tzinfo: value = value.astimezone(timezone.utc).replace(tzinfo=None) return value.strftime("%Y-%m-%d %H:%M:%S") if isinstance(value, date): return value.isoformat() raise TypeError(f"{type(value).__name__} is not JSON serializable") def log_event(event_name, **fields): print( json.dumps( {"event": event_name, **fields}, default=json_default, separators=(",", ":"), sort_keys=True, ) ) def secrets_client(): global _secrets_client if _secrets_client is None: _secrets_client = boto3.client("secretsmanager") return _secrets_client def load_service_account_info(): secret_arn = env("GCP_SERVICE_ACCOUNT_SECRET_ARN", required=True) response = secrets_client().get_secret_value(SecretId=secret_arn) secret_string = response.get("SecretString") if not secret_string: raise RuntimeError("GCP service account secret must use SecretString") try: info = json.loads(secret_string) except json.JSONDecodeError as exc: raise RuntimeError("GCP service account secret is not valid JSON") from exc if info.get("type") != "service_account": raise RuntimeError("GCP service account secret has an invalid type") if not info.get("client_email") or not info.get("private_key"): raise RuntimeError("GCP service account secret is missing required fields") return info def bigquery_session(service_account_info): from google.oauth2 import service_account from google.auth.transport.requests import AuthorizedSession credentials = service_account.Credentials.from_service_account_info( service_account_info, scopes=["https://www.googleapis.com/auth/bigquery"], ) return AuthorizedSession(credentials) def mysql_connect_url(database_url, autocommit=False): parsed = urllib.parse.urlparse(database_url) if not parsed.hostname or not parsed.path.lstrip("/"): raise RuntimeError("database URL must include host and database") query = urllib.parse.parse_qs(parsed.query) ssl_ctx = ssl.create_default_context() ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2 ca_path = env("TARGET_TIDB_SSL_CA_PATH") or env("TIDB_SSL_CA_PATH") if ca_path: ssl_ctx.load_verify_locations(cafile=ca_path) return pymysql.connect( host=parsed.hostname, port=parsed.port or 4000, user=urllib.parse.unquote(parsed.username or ""), password=urllib.parse.unquote(parsed.password or ""), database=urllib.parse.unquote(parsed.path.lstrip("/")), charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=autocommit, connect_timeout=int((query.get("connect_timeout") or ["10"])[0]), read_timeout=int((query.get("read_timeout") or ["60"])[0]), write_timeout=int((query.get("write_timeout") or ["60"])[0]), ssl=ssl_ctx, ) def devops_db_connect(): return pymysql.connect( host=env("DEVOPS_DB_HOST", required=True), port=int(env("DEVOPS_DB_PORT", "3306")), user=env("DEVOPS_DB_USER", required=True), password=env("DEVOPS_DB_PASSWORD", required=True), database=env("DEVOPS_DB_NAME", required=True), charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=True, connect_timeout=10, read_timeout=60, write_timeout=60, ) def target_db_connect(): return mysql_connect_url(env("TARGET_DATABASE_URL", required=True)) def parse_bool(value, default=False): if value is None: return default if isinstance(value, bool): return value return str(value).strip().lower() not in {"0", "false", "no", "off", ""} def validate_tenant_ids(values): if not isinstance(values, (list, tuple, set)): raise ValueError("tenant_ids must be an array") tenant_ids = [] seen = set() for value in values or []: tenant_id = str(value).strip() if not TENANT_ID_PATTERN.fullmatch(tenant_id): raise ValueError("tenant_ids must contain numeric IDs only") if tenant_id not in seen: seen.add(tenant_id) tenant_ids.append(tenant_id) maximum = int(env("OWNER_SYNC_MAX_TENANTS", "20000")) if len(tenant_ids) > maximum: raise ValueError(f"tenant count exceeds OWNER_SYNC_MAX_TENANTS={maximum}") return tenant_ids def select_full_tenant_ids(): connection = devops_db_connect() try: selected = [] with connection.cursor() as cursor: for sql in (ACTIVE_TENANT_IDS_SQL, NEW_SIGNUP_TENANT_IDS_SQL): cursor.execute(sql) selected.extend(row["id"] for row in cursor.fetchall()) return validate_tenant_ids(selected) finally: connection.close() def read_bigquery_sql(): return Path(__file__).with_name("effective_sales_owner.sql").read_text( encoding="utf-8" ) def bigquery_value(value, field_type): if value is None: return None if field_type == "BOOL": return str(value).lower() == "true" if field_type in {"INTEGER", "INT64"}: return int(value) if field_type in {"FLOAT", "FLOAT64"}: return float(value) if field_type == "TIMESTAMP": try: return datetime.fromtimestamp(float(value), tz=timezone.utc) except (TypeError, ValueError): return datetime.fromisoformat(str(value).replace("Z", "+00:00")) return value def decode_bigquery_rows(response): fields = (response.get("schema") or {}).get("fields") or [] decoded = [] for raw_row in response.get("rows") or []: cells = raw_row.get("f") or [] decoded.append( { field["name"]: bigquery_value( cells[index].get("v") if index < len(cells) else None, field.get("type", "STRING"), ) for index, field in enumerate(fields) } ) return decoded def query_owner_candidates(session, tenant_ids): project = env("BIGQUERY_JOB_PROJECT", "gcp-resilience-dev") endpoint = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project}/queries" request_timeout = int(env("BIGQUERY_HTTP_TIMEOUT_SECONDS", "60")) payload = { "query": read_bigquery_sql(), "useLegacySql": False, "parameterMode": "NAMED", "queryParameters": [ { "name": "tenant_ids", "parameterType": { "type": "ARRAY", "arrayType": {"type": "STRING"}, }, "parameterValue": { "arrayValues": [{"value": value} for value in tenant_ids] }, } ], "timeoutMs": min(50000, max(1000, (request_timeout - 5) * 1000)), "maxResults": 10000, } response = session.post(endpoint, json=payload, timeout=request_timeout) response.raise_for_status() page = response.json() job_reference = page.get("jobReference") or {} job_id = job_reference.get("jobId") location = job_reference.get("location") deadline = time.monotonic() + int(env("BIGQUERY_QUERY_TIMEOUT_SECONDS", "720")) rows = [] schema = page.get("schema") while True: if page.get("errors"): raise RuntimeError("BigQuery query job failed") if page.get("jobComplete"): if schema and not page.get("schema"): page["schema"] = schema schema = page.get("schema") or schema rows.extend(decode_bigquery_rows(page)) page_token = page.get("pageToken") if not page_token: return rows else: page_token = None if not job_id or time.monotonic() >= deadline: raise TimeoutError("BigQuery query did not complete before timeout") params = {"maxResults": 10000, "timeoutMs": 10000} if location: params["location"] = location if page_token: params["pageToken"] = page_token result_url = ( f"https://bigquery.googleapis.com/bigquery/v2/projects/{project}" f"/queries/{job_id}" ) response = session.get(result_url, params=params, timeout=request_timeout) response.raise_for_status() page = response.json() def normalize_datetime(value): if value is None: return None if isinstance(value, datetime) and value.tzinfo: return value.astimezone(timezone.utc).replace(tzinfo=None) return value def normalized_email(value): return str(value or "").strip().lower() or None def build_snapshot_row(row, sync_id, snapshot_at): account_owner_email = normalized_email(row.get("account_owner_email")) opportunity_owner_email = normalized_email(row.get("opportunity_owner_email")) if row.get("account_owner_active") and account_owner_email: effective_owner_id = row.get("account_owner_id") effective_owner_email = account_owner_email effective_owner_source = "account" elif row.get("opportunity_owner_active") and opportunity_owner_email: effective_owner_id = row.get("opportunity_owner_id") effective_owner_email = opportunity_owner_email effective_owner_source = "opportunity" else: effective_owner_id = None effective_owner_email = None effective_owner_source = "none" if effective_owner_source != "none": decision_status = "owner_selected" elif not row.get("org_matched"): decision_status = "no_crm_org" else: decision_status = "no_active_owner" return { "sync_id": sync_id, "tenant_id": str(row["tenant_id"]), "org_matched": bool(row.get("org_matched")), "account_id": row.get("account_id"), "account_matched": bool(row.get("account_matched")), "account_approved": bool(row.get("account_approved")), "account_owner_id": row.get("account_owner_id"), "account_owner_active": bool(row.get("account_owner_active")), "lead_id": row.get("lead_id"), "lead_owner_id": row.get("lead_owner_id"), "lead_owner_active": bool(row.get("lead_owner_active")), "selected_opportunity_id": row.get("selected_opportunity_id"), "opportunity_stage": row.get("opportunity_stage"), "opportunity_has_open_activity": bool( row.get("opportunity_has_open_activity") ), "opportunity_owner_id": row.get("opportunity_owner_id"), "opportunity_owner_active": bool(row.get("opportunity_owner_active")), "effective_owner_id": effective_owner_id, "effective_owner_email": effective_owner_email, "effective_owner_source": effective_owner_source, "effective_owner_active": effective_owner_source != "none", "decision_status": decision_status, "rule_version": RULE_VERSION, "source_read_at": normalize_datetime(row.get("source_read_at")), "snapshot_synced_at": snapshot_at, } def aggregate_counts(rows): return { "tenant_count": len(rows), "owner_selected_count": sum( row["effective_owner_source"] != "none" for row in rows ), "account_owner_count": sum( row["effective_owner_source"] == "account" for row in rows ), "opportunity_owner_count": sum( row["effective_owner_source"] == "opportunity" for row in rows ), "no_owner_count": sum( row["effective_owner_source"] == "none" for row in rows ), } def record_attempt(sync_id, status, error_code=None): connection = target_db_connect() try: with connection.cursor() as cursor: cursor.execute( """ INSERT INTO apac_effective_sales_owner_sync_state ( state_id, last_attempt_sync_id, last_attempt_at, last_attempt_status, last_error_code ) VALUES (%s, %s, UTC_TIMESTAMP(), %s, %s) ON DUPLICATE KEY UPDATE last_attempt_sync_id = VALUES(last_attempt_sync_id), last_attempt_at = VALUES(last_attempt_at), last_attempt_status = VALUES(last_attempt_status), last_error_code = VALUES(last_error_code) """, (STATE_ID, sync_id, status, error_code), ) connection.commit() finally: connection.close() def persist_snapshot(sync_id, rows, counts, snapshot_at): connection = target_db_connect() placeholders = ",".join(["%s"] * len(SNAPSHOT_COLUMNS)) columns = ",".join(SNAPSHOT_COLUMNS) sql = ( f"INSERT INTO apac_effective_sales_owner_snapshot ({columns}) " f"VALUES ({placeholders})" ) batch_size = int(env("OWNER_SYNC_DB_BATCH_SIZE", "500")) try: with connection.cursor() as cursor: for start in range(0, len(rows), batch_size): batch = rows[start : start + batch_size] cursor.executemany( sql, [tuple(row[column] for column in SNAPSHOT_COLUMNS) for row in batch], ) owner_updates = [ ( row["effective_owner_email"], row["effective_owner_source"], row["tenant_id"], ) for row in rows ] for table_name in ( "apac_active_tenants", "apac_sales_owned_tenants", ): cursor.executemany( f""" UPDATE {table_name} SET effective_sales_owner_email = %s, effective_sales_owner_source = %s WHERE tenant_id = %s """, owner_updates, ) cursor.execute( """ INSERT INTO apac_effective_sales_owner_sync_state ( state_id, last_successful_sync_id, last_successful_sync_at, last_successful_tenant_count, last_successful_owner_count, last_successful_rule_version, last_attempt_sync_id, last_attempt_at, last_attempt_status, last_error_code ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, 'succeeded', NULL) ON DUPLICATE KEY UPDATE last_successful_sync_id = VALUES(last_successful_sync_id), last_successful_sync_at = VALUES(last_successful_sync_at), last_successful_tenant_count = VALUES(last_successful_tenant_count), last_successful_owner_count = VALUES(last_successful_owner_count), last_successful_rule_version = VALUES(last_successful_rule_version), last_attempt_sync_id = VALUES(last_attempt_sync_id), last_attempt_at = VALUES(last_attempt_at), last_attempt_status = VALUES(last_attempt_status), last_error_code = NULL """, ( STATE_ID, sync_id, snapshot_at, counts["tenant_count"], counts["owner_selected_count"], RULE_VERSION, sync_id, snapshot_at, ), ) retention_days = int(env("OWNER_SYNC_RETENTION_DAYS", "14")) retention_cutoff = snapshot_at - timedelta(days=retention_days) cursor.execute( """ DELETE FROM apac_effective_sales_owner_snapshot WHERE snapshot_synced_at < %s AND sync_id <> %s """, (retention_cutoff, sync_id), ) connection.commit() except Exception: connection.rollback() raise finally: connection.close() def lambda_handler(event, context): event = event or {} started = time.time() sync_id = str(event.get("sync_id") or uuid.uuid4()) if not SYNC_ID_PATTERN.fullmatch(sync_id): raise ValueError("sync_id has an invalid format") dry_run = parse_bool(event.get("dry_run"), False) supplied_tenant_ids = event.get("tenant_ids") mode = "selected" if supplied_tenant_ids is not None else "full" log_event("effective_owner_sync_started", sync_id=sync_id, mode=mode, dry_run=dry_run) try: tenant_ids = ( validate_tenant_ids(supplied_tenant_ids) if supplied_tenant_ids is not None else select_full_tenant_ids() ) if not tenant_ids: raise RuntimeError("owner sync selected zero tenants") if not dry_run: record_attempt(sync_id, "running") service_account_info = load_service_account_info() try: candidates = query_owner_candidates( bigquery_session(service_account_info), tenant_ids ) finally: service_account_info.clear() if len(candidates) != len(tenant_ids): raise RuntimeError("BigQuery did not return exactly one row per tenant") snapshot_at = datetime.now(timezone.utc).replace(tzinfo=None) rows = [ build_snapshot_row(row, sync_id, snapshot_at) for row in candidates ] counts = aggregate_counts(rows) if not dry_run: persist_snapshot(sync_id, rows, counts, snapshot_at) elapsed = round(time.time() - started, 3) log_event( "effective_owner_sync_completed", sync_id=sync_id, mode=mode, dry_run=dry_run, rule_version=RULE_VERSION, elapsed_seconds=elapsed, **counts, ) return { "sync_id": sync_id, "mode": mode, "dry_run": dry_run, "rule_version": RULE_VERSION, "elapsed_seconds": elapsed, **counts, } except Exception as exc: error_code = type(exc).__name__ if not dry_run: try: record_attempt(sync_id, "failed", error_code) except Exception as state_exc: log_event( "effective_owner_sync_state_update_failed", sync_id=sync_id, error_code=type(state_exc).__name__, ) log_event("effective_owner_sync_failed", sync_id=sync_id, error_code=error_code) raise RuntimeError(f"effective owner sync failed: {error_code}") from None