#!/usr/bin/env python3 from __future__ import annotations import hmac import base64 import hashlib import json import os import re import sqlite3 import threading import time from dataclasses import dataclass from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import parse_qs, quote, unquote, urlparse from urllib.request import Request, urlopen import common API_VERSION = "v63.0" DEFAULT_LIMIT = 100 CAMPAIGN_SEARCH_SCAN_LIMIT = 1000 ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,128}$") SALESFORCE_ID_PATTERN = re.compile(r"^[A-Za-z0-9]{15}(?:[A-Za-z0-9]{3})?$") MIN_OPTIONS_TOKEN_LENGTH = 16 WEAK_OPTIONS_TOKENS = { "changeme", "dummy", "example", "password", "placeholder", "secret", "test", "token", } WEAK_OPTIONS_TOKEN_PREFIXES = WEAK_OPTIONS_TOKENS - {"token"} | {"optionstoken"} DEFAULT_WORKFLOW_ACTION_REPLAY_RPS = 80.0 DEFAULT_WORKFLOW_ACTION_MAX_ATTEMPTS = 12 # At the capped five-minute backoff these reserve about four days for CRM # completion and two days for callback delivery inside HubSpot's one-week block. DEFAULT_WORKFLOW_ACTION_RESULT_MAX_ATTEMPTS = 1152 DEFAULT_WORKFLOW_ACTION_CALLBACK_MAX_ATTEMPTS = 576 DEFAULT_WORKFLOW_ACTION_ADAPTER_BACKLOG = 1024 WORKFLOW_ACTION_PROCESSING_LEASE_SECONDS = 120.0 WORKFLOW_ACTION_POLICIES = {"best_effort", "legacy_replay", "must_complete"} HUBSPOT_SIGNATURE_SKEW_MS = 5 * 60 * 1000 WORKFLOW_ACTION_WEBHOOK_PATH = "/webhooks/hubspot-crm-connector/workflow-action" WORKFLOW_ACTION_EXECUTE_PATH = "/hubspot/workflow-actions/execute" WORKFLOW_ACTION_CALLBACK_BASE_URL = "https://api.hubapi.com/automation/actions/callbacks/2026-03" class AdapterError(Exception): def __init__(self, status: HTTPStatus, message: str): super().__init__(message) self.status = status self.message = message def retryable_http_status(status: int) -> bool: return status in { HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_EARLY, HTTPStatus.TOO_MANY_REQUESTS, } or status >= HTTPStatus.INTERNAL_SERVER_ERROR @dataclass(frozen=True) class AdapterConfig: listen_host: str listen_port: int backend_base_url: str options_token: str username: str password: str client_id: str client_secret: str hubspot_client_secret: str hubspot_access_token: str workflow_action_callback_access_token: str workflow_action_queue_db: str workflow_action_replay_url: str workflow_action_replay_signing_url: str workflow_action_replay_signing_secret: str workflow_action_replay_rps: float workflow_action_max_attempts: int workflow_action_result_max_attempts: int workflow_action_callback_max_attempts: int workflow_action_adapter_backlog: int workflow_action_execute_mode: str workflow_action_policies: dict[str, str] runtime_env: dict[str, str] def env_any(envv: dict[str, str], *names: str, default: str = "") -> str: for name in names: value = envv.get(name, "").strip() if value: return value return default def parse_port(raw: str, label: str) -> int: value = str(raw).strip() if not value.isdigit(): raise RuntimeError(f"{label} must be a TCP port") port = int(value) if port < 1 or port > 65535: raise RuntimeError(f"{label} must be a TCP port") return port def parse_positive_float(raw: str, label: str) -> float: try: value = float(str(raw).strip()) except ValueError as exc: raise RuntimeError(f"{label} must be a positive number") from exc if value <= 0: raise RuntimeError(f"{label} must be a positive number") return value def parse_positive_int(raw: str, label: str) -> int: value = str(raw).strip() if not value.isdigit() or int(value) <= 0: raise RuntimeError(f"{label} must be a positive integer") return int(value) def parse_workflow_action_policies(raw: str) -> dict[str, str]: if not raw.strip(): return {} try: value = json.loads(raw) except json.JSONDecodeError as exc: raise RuntimeError("EXP_HUBSPOT_WORKFLOW_ACTION_POLICIES_JSON must be a JSON object") from exc if not isinstance(value, dict) or not value: raise RuntimeError("EXP_HUBSPOT_WORKFLOW_ACTION_POLICIES_JSON must be a non-empty JSON object") policies: dict[str, str] = {} for raw_action_id, raw_policy in value.items(): action_id = str(raw_action_id).strip() policy = str(raw_policy).strip().lower() if not action_id or action_id == "*": raise RuntimeError("workflow action policy keys must be exact actionDefinitionId values") if policy not in WORKFLOW_ACTION_POLICIES: raise RuntimeError(f"unsupported workflow action policy for {action_id}: {policy}") policies[action_id] = policy return policies def validate_options_token(token: str) -> None: normalized = re.sub(r"[^a-z0-9]+", "", token.lower()) if normalized in WEAK_OPTIONS_TOKENS or any( normalized.startswith(prefix) for prefix in WEAK_OPTIONS_TOKEN_PREFIXES ): raise RuntimeError("EXP_HUBSPOT_ACTION_OPTIONS_TOKEN must not be a placeholder value") if len(token) < MIN_OPTIONS_TOKEN_LENGTH: raise RuntimeError(f"EXP_HUBSPOT_ACTION_OPTIONS_TOKEN must be at least {MIN_OPTIONS_TOKEN_LENGTH} characters") def load_config(envv: dict[str, str] | None = None) -> AdapterConfig: envv = envv or os.environ backend_port = envv.get("BACKEND_PORT", "").strip() adapter_port = envv.get("EXP_HUBSPOT_OPTIONS_ADAPTER_PORT", "").strip() if not adapter_port: adapter_port = str(parse_port(backend_port, "BACKEND_PORT") + 101) options_token = envv.get("EXP_HUBSPOT_ACTION_OPTIONS_TOKEN", "").strip() password = envv.get("NEXUS_ADMIN_PASSWORD", "").strip() if not options_token: raise RuntimeError("EXP_HUBSPOT_ACTION_OPTIONS_TOKEN is required") validate_options_token(options_token) if not password: raise RuntimeError("NEXUS_ADMIN_PASSWORD is required") backend_base_url = envv.get("NEXUSCRM_BACKEND_BASE_URL", f"http://127.0.0.1:{backend_port}").rstrip("/") env_root = envv.get("ENV_ROOT", "").strip() default_queue_db = ( os.path.join(env_root, "data", "hubspot-workflow-action-queue.sqlite3") if env_root else "hubspot-workflow-action-queue.sqlite3" ) public_base_url = envv.get("PUBLIC_BASE_URL", "").strip().rstrip("/") replay_url = envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_URL", "").strip() or f"{backend_base_url}{WORKFLOW_ACTION_WEBHOOK_PATH}" default_replay_signing_url = f"{public_base_url}{WORKFLOW_ACTION_WEBHOOK_PATH}" if public_base_url else replay_url replay_signing_url = ( envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_SIGNING_URL", "").strip() or envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_URL", "").strip() or default_replay_signing_url ) workflow_action_execute_mode = envv.get("HUBSPOT_WORKFLOW_EXECUTE_MODE", "block").strip().lower() or "block" if workflow_action_execute_mode not in {"block", "proxy"}: raise RuntimeError("HUBSPOT_WORKFLOW_EXECUTE_MODE must be one of block, proxy") hubspot_client_secret = envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_SIGNING_SECRET", "").strip() workflow_action_policies = ( parse_workflow_action_policies(envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_POLICIES_JSON", "")) if workflow_action_execute_mode == "proxy" else {} ) if workflow_action_execute_mode == "proxy" and not hubspot_client_secret: raise RuntimeError( "EXP_HUBSPOT_WORKFLOW_ACTION_SIGNING_SECRET is required when workflow action execution is proxied" ) if workflow_action_execute_mode == "proxy" and not workflow_action_policies: raise RuntimeError( "EXP_HUBSPOT_WORKFLOW_ACTION_POLICIES_JSON is required when workflow action execution is proxied" ) hubspot_access_token = env_any( envv, "EXP_HUBSPOT_ACCESS_TOKEN", "HUBSPOT_ACCESS_TOKEN" ) workflow_action_callback_access_token = envv.get( "EXP_HUBSPOT_WORKFLOW_ACTION_CALLBACK_ACCESS_TOKEN", "" ).strip() workflow_action_replay_signing_secret = env_any( envv, "EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_SIGNING_SECRET", "EXP_HUBSPOT_CLIENT_SECRET", "HUBSPOT_CLIENT_SECRET", ) if "must_complete" in workflow_action_policies.values() and not workflow_action_callback_access_token: raise RuntimeError( "EXP_HUBSPOT_WORKFLOW_ACTION_CALLBACK_ACCESS_TOKEN is required for must_complete workflow actions" ) if "must_complete" in workflow_action_policies.values(): required_result_store_keys = [ "TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_PASSWORD", "TIDB_DATABASE", ] missing_result_store_keys = [ key for key in required_result_store_keys if not envv.get(key, "").strip() ] if missing_result_store_keys: raise RuntimeError( "must_complete workflow actions require Connector Event result-store access: " + ", ".join(missing_result_store_keys) ) if {"legacy_replay", "must_complete"} & set(workflow_action_policies.values()): if not hubspot_access_token: raise RuntimeError( "EXP_HUBSPOT_ACCESS_TOKEN is required for replayed workflow actions" ) if not workflow_action_replay_signing_secret: raise RuntimeError( "workflow action replay signing secret is required for replayed workflow actions" ) return AdapterConfig( listen_host=envv.get("EXP_HUBSPOT_OPTIONS_ADAPTER_HOST", "127.0.0.1").strip() or "127.0.0.1", listen_port=parse_port(adapter_port, "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT"), backend_base_url=backend_base_url, options_token=options_token, username=env_any(envv, "EXP_NEXUS_ADMIN_USERNAME", "NEXUS_ADMIN_USERNAME", "VERIFY_USERNAME", default="admin@example.invalid"), password=password, client_id=env_any(envv, "EXP_NEXUS_CLIENT_ID", "NEXUS_CLIENT_ID", default="nexuscrm-ui"), client_secret=env_any(envv, "EXP_NEXUS_CLIENT_SECRET", "NEXUS_CLIENT_SECRET", default="nexuscrm-ui-internal"), hubspot_client_secret=hubspot_client_secret, hubspot_access_token=hubspot_access_token, workflow_action_callback_access_token=workflow_action_callback_access_token, workflow_action_replay_signing_secret=workflow_action_replay_signing_secret, workflow_action_queue_db=envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_QUEUE_DB", default_queue_db).strip() or default_queue_db, workflow_action_replay_url=replay_url, workflow_action_replay_signing_url=replay_signing_url, workflow_action_replay_rps=parse_positive_float( envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_RPS", str(DEFAULT_WORKFLOW_ACTION_REPLAY_RPS)), "EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_RPS", ), workflow_action_max_attempts=parse_positive_int( envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_MAX_ATTEMPTS", str(DEFAULT_WORKFLOW_ACTION_MAX_ATTEMPTS)), "EXP_HUBSPOT_WORKFLOW_ACTION_MAX_ATTEMPTS", ), workflow_action_result_max_attempts=parse_positive_int( envv.get( "EXP_HUBSPOT_WORKFLOW_ACTION_RESULT_MAX_ATTEMPTS", str(DEFAULT_WORKFLOW_ACTION_RESULT_MAX_ATTEMPTS), ), "EXP_HUBSPOT_WORKFLOW_ACTION_RESULT_MAX_ATTEMPTS", ), workflow_action_callback_max_attempts=parse_positive_int( envv.get( "EXP_HUBSPOT_WORKFLOW_ACTION_CALLBACK_MAX_ATTEMPTS", str(DEFAULT_WORKFLOW_ACTION_CALLBACK_MAX_ATTEMPTS), ), "EXP_HUBSPOT_WORKFLOW_ACTION_CALLBACK_MAX_ATTEMPTS", ), workflow_action_adapter_backlog=parse_positive_int( envv.get("EXP_HUBSPOT_WORKFLOW_ACTION_ADAPTER_BACKLOG", str(DEFAULT_WORKFLOW_ACTION_ADAPTER_BACKLOG)), "EXP_HUBSPOT_WORKFLOW_ACTION_ADAPTER_BACKLOG", ), workflow_action_execute_mode=workflow_action_execute_mode, workflow_action_policies=workflow_action_policies, runtime_env=dict(envv), ) def json_response( handler: BaseHTTPRequestHandler, status: HTTPStatus, payload: dict[str, Any], *, headers: dict[str, str] | None = None, ) -> None: body = json.dumps(payload, separators=(",", ":")).encode("utf-8") handler.send_response(status.value) handler.send_header("Content-Type", "application/json") handler.send_header("Cache-Control", "no-store") for key, value in (headers or {}).items(): handler.send_header(key, value) handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body) def read_raw_body(handler: BaseHTTPRequestHandler) -> bytes: length = int(handler.headers.get("Content-Length") or "0") if length <= 0: return b"" return handler.rfile.read(min(length, 1024 * 1024)) def parse_json_body(raw: bytes) -> dict[str, Any]: if not raw: return {} try: parsed = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as exc: raise AdapterError(HTTPStatus.BAD_REQUEST, "invalid JSON body") from exc if not isinstance(parsed, dict): raise AdapterError(HTTPStatus.BAD_REQUEST, "JSON body must be an object") return parsed def read_json_body(handler: BaseHTTPRequestHandler) -> dict[str, Any]: return parse_json_body(read_raw_body(handler)) def workflow_action_definition_id(body: dict[str, Any]) -> str: origin = body.get("origin") if not isinstance(origin, dict): raise AdapterError(HTTPStatus.UNPROCESSABLE_ENTITY, "workflow action origin.actionDefinitionId is required") value = origin.get("actionDefinitionId") if value is None or not str(value).strip(): raise AdapterError(HTTPStatus.UNPROCESSABLE_ENTITY, "workflow action origin.actionDefinitionId is required") return str(value).strip() def workflow_action_policy(config: AdapterConfig, body: dict[str, Any]) -> str: action_definition_id = workflow_action_definition_id(body) policy = config.workflow_action_policies.get(action_definition_id) if policy is None: raise AdapterError( HTTPStatus.UNPROCESSABLE_ENTITY, f"workflow action {action_definition_id} has no configured delivery policy", ) return policy def token_from_request(query: dict[str, list[str]], body: dict[str, Any]) -> str: for key in ["options_token", "token"]: if query.get(key): return query[key][0] for key in ["options_token", "token"]: value = body.get(key) if isinstance(value, str): return value return "" def verify_options_token(config: AdapterConfig, query: dict[str, list[str]], body: dict[str, Any]) -> None: provided = token_from_request(query, body) if not hmac.compare_digest(provided, config.options_token): raise AdapterError(HTTPStatus.UNAUTHORIZED, "invalid options token") def search_text(query: dict[str, list[str]], body: dict[str, Any]) -> str: for key in ["q", "query", "search", "searchQuery"]: if query.get(key): return query[key][0].strip() value = body.get(key) if isinstance(value, str) and value.strip(): return value.strip() fetch_options = body.get("fetchOptions") if isinstance(fetch_options, dict): value = fetch_options.get("q") or fetch_options.get("query") or fetch_options.get("searchQuery") if isinstance(value, str): return value.strip() return "" def selected_option_value(query: dict[str, list[str]], body: dict[str, Any]) -> str: for key in ["value", "selectedValue", "currentValue", "initialValue"]: if query.get(key): return query[key][0].strip() value = body.get(key) if isinstance(value, str) and value.strip(): return value.strip() fetch_options = body.get("fetchOptions") if isinstance(fetch_options, dict): value = ( fetch_options.get("value") or fetch_options.get("selectedValue") or fetch_options.get("currentValue") or fetch_options.get("initialValue") ) if isinstance(value, str): return value.strip() return "" def pagination_after(query: dict[str, list[str]], body: dict[str, Any]) -> int: raw: Any = "" if query.get("after"): raw = query["after"][0] elif isinstance(body.get("after"), (int, str)): raw = body.get("after") else: fetch_options = body.get("fetchOptions") if isinstance(fetch_options, dict): raw = fetch_options.get("after") or "" try: offset = int(str(raw).strip() or "0") except ValueError: return 0 return max(offset, 0) def input_field_value(body: dict[str, Any], *names: str) -> str: input_fields = body.get("inputFields") if not isinstance(input_fields, dict): return "" for name in names: field = input_fields.get(name) if isinstance(field, dict): value = field.get("value") if isinstance(value, str) and value.strip(): return value.strip() if isinstance(field, str) and field.strip(): return field.strip() return "" def campaign_id_from_request(query: dict[str, list[str]], body: dict[str, Any]) -> str: for key in ["campaign_id", "campaignId", "CampaignId"]: if query.get(key): return query[key][0].strip() value = body.get(key) if isinstance(value, str) and value.strip(): return value.strip() return input_field_value(body, "campaign_id", "CampaignId") def escape_soql_string(value: str) -> str: return value.replace("\\", "\\\\").replace("'", "\\'") def salesforce_id_filter(value: str) -> str: escaped = escape_soql_string(value) if len(value) == 15: return f"Id LIKE '{escaped}%'" return f"Id = '{escaped}'" def oauth_password_grant(config: AdapterConfig) -> str: payload = { "grant_type": "password", "client_id": config.client_id, "client_secret": config.client_secret, "username": config.username, "password": config.password, } encoded = "&".join(f"{quote(key)}={quote(value)}" for key, value in payload.items()).encode("utf-8") request = Request( f"{config.backend_base_url}/services/oauth2/token", data=encoded, method="POST", headers={"Content-Type": "application/x-www-form-urlencoded"}, ) try: with urlopen(request, timeout=15) as response: body = json.loads(response.read().decode("utf-8")) except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as exc: raise AdapterError(HTTPStatus.BAD_GATEWAY, "failed to authenticate with NexusCRM") from exc token = str(body.get("access_token", "")).strip() if not token: raise AdapterError(HTTPStatus.BAD_GATEWAY, "NexusCRM token response is missing access_token") return token def nexus_query(config: AdapterConfig, access_token: str, soql: str) -> list[dict[str, Any]]: path = f"/services/data/{API_VERSION}/query?q={quote(soql, safe='')}" request = Request( f"{config.backend_base_url}{path}", method="GET", headers={"Authorization": f"Bearer {access_token}"}, ) try: with urlopen(request, timeout=20) as response: body = json.loads(response.read().decode("utf-8")) except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as exc: raise AdapterError(HTTPStatus.BAD_GATEWAY, "failed to query NexusCRM") from exc records = body.get("records", []) if not isinstance(records, list): raise AdapterError(HTTPStatus.BAD_GATEWAY, "NexusCRM query response is missing records") return [record for record in records if isinstance(record, dict)] def option_payload(searchable: bool, options: list[dict[str, str]], after: int | None = None) -> dict[str, Any]: payload: dict[str, Any] = {"searchable": searchable, "options": options} if after is not None: payload["after"] = str(after) return payload def campaign_option(record: dict[str, Any]) -> dict[str, str] | None: campaign_id = str(record.get("Id") or "") if not campaign_id: return None return {"label": str(record.get("Name") or campaign_id), "value": campaign_id} def campaign_search_rank(name: str, term: str) -> tuple[int, str]: normalized_name = name.casefold() normalized_term = term.casefold() if normalized_name == normalized_term: return (0, normalized_name) if normalized_name.startswith(normalized_term): return (1, normalized_name) tokenized = re.sub(r"[^a-z0-9]+", " ", normalized_name).split() if any(token.startswith(normalized_term) for token in tokenized): return (2, normalized_name) return (3, normalized_name) def campaign_options_from_records(records: list[dict[str, Any]]) -> list[dict[str, str]]: options: list[dict[str, str]] = [] seen: set[str] = set() for record in records: option = campaign_option(record) if option is None or option["value"] in seen: continue seen.add(option["value"]) options.append(option) return options def campaign_options(config: AdapterConfig, query: dict[str, list[str]], body: dict[str, Any]) -> dict[str, Any]: token = oauth_password_grant(config) search_term = search_text(query, body) selected_value = selected_option_value(query, body) or input_field_value(body, "campaign_id", "CampaignId") if not search_term and selected_value and ID_PATTERN.match(selected_value): soql = ( "SELECT Id, Name FROM Campaign " f"WHERE IsDeleted = false AND IsActive = true AND {salesforce_id_filter(selected_value)} " "ORDER BY Name ASC LIMIT 1" ) records = nexus_query(config, token, soql) return option_payload(True, campaign_options_from_records(records)) after = pagination_after(query, body) if search_term: escaped_term = escape_soql_string(search_term) search_filter = f"Name LIKE '%{escaped_term}%'" if SALESFORCE_ID_PATTERN.match(search_term): search_filter = f"({search_filter} OR {salesforce_id_filter(search_term)})" soql = ( "SELECT Id, Name FROM Campaign " f"WHERE IsDeleted = false AND IsActive = true AND {search_filter} " f"ORDER BY Name ASC LIMIT {CAMPAIGN_SEARCH_SCAN_LIMIT}" ) records = nexus_query(config, token, soql) records.sort(key=lambda record: campaign_search_rank(str(record.get("Name") or ""), search_term)) options = campaign_options_from_records(records) page = options[after : after + DEFAULT_LIMIT] next_after = after + DEFAULT_LIMIT if after + DEFAULT_LIMIT < len(options) else None return option_payload(True, page, next_after) soql = ( "SELECT Id, Name FROM Campaign " "WHERE IsDeleted = false AND IsActive = true " f"ORDER BY Name ASC LIMIT {DEFAULT_LIMIT + 1} OFFSET {after}" ) records = nexus_query(config, token, soql) options = campaign_options_from_records(records[:DEFAULT_LIMIT]) next_after = after + DEFAULT_LIMIT if len(records) > DEFAULT_LIMIT else None return option_payload(True, options, next_after) def status_options(config: AdapterConfig, query: dict[str, list[str]], body: dict[str, Any]) -> dict[str, Any]: campaign_id = campaign_id_from_request(query, body) if not campaign_id: return {"searchable": False, "options": []} if not ID_PATTERN.match(campaign_id): raise AdapterError(HTTPStatus.BAD_REQUEST, "invalid campaign_id") term = search_text(query, body) or selected_option_value(query, body) filters = [f"CampaignId = '{escape_soql_string(campaign_id)}'"] if term: filters.append(f"Label LIKE '%{escape_soql_string(term)}%'") soql = ( "SELECT Label, SortOrder, IsDefault FROM CampaignMemberStatus " f"WHERE {' AND '.join(filters)} " f"ORDER BY SortOrder ASC, Label ASC LIMIT {DEFAULT_LIMIT}" ) token = oauth_password_grant(config) records = nexus_query(config, token, soql) return { "searchable": False, "options": [ { "label": str(record.get("Label") or ""), "value": str(record.get("Label") or ""), "description": "Default" if record.get("IsDefault") else "", } for record in records if record.get("Label") ], } def request_url(handler: BaseHTTPRequestHandler) -> str: proto = handler.headers.get("X-Forwarded-Proto") or "http" host = handler.headers.get("Host") or f"{handler.server.server_address[0]}:{handler.server.server_address[1]}" return f"{proto}://{host}{handler.path}" def verify_hubspot_v3_signature(config: AdapterConfig, method: str, url: str, raw_body: bytes, headers: Any) -> None: if not config.hubspot_client_secret: raise AdapterError(HTTPStatus.SERVICE_UNAVAILABLE, "workflow action signing secret is not configured") timestamp = str(headers.get("X-HubSpot-Request-Timestamp") or "").strip() signature = str(headers.get("X-HubSpot-Signature-v3") or "").strip() if not timestamp or not signature: raise AdapterError(HTTPStatus.UNAUTHORIZED, "missing HubSpot signature") try: timestamp_ms = int(timestamp) except ValueError as exc: raise AdapterError(HTTPStatus.UNAUTHORIZED, "invalid HubSpot timestamp") from exc if abs(int(time.time() * 1000) - timestamp_ms) > HUBSPOT_SIGNATURE_SKEW_MS: raise AdapterError(HTTPStatus.UNAUTHORIZED, "stale HubSpot timestamp") source = method.upper().encode("utf-8") + url.encode("utf-8") + raw_body + timestamp.encode("utf-8") digest = hmac.new(config.hubspot_client_secret.encode("utf-8"), source, hashlib.sha256).digest() expected = base64.b64encode(digest).decode("ascii") if not hmac.compare_digest(expected, signature): raise AdapterError(HTTPStatus.UNAUTHORIZED, "invalid HubSpot signature") def hubspot_signature_headers(secret: str, method: str, signing_url: str, raw_body: bytes) -> dict[str, str]: timestamp = str(int(time.time() * 1000)) source = method.upper().encode("utf-8") + signing_url.encode("utf-8") + raw_body + timestamp.encode("utf-8") digest = hmac.new(secret.encode("utf-8"), source, hashlib.sha256).digest() return { "X-HubSpot-Request-Timestamp": timestamp, "X-HubSpot-Signature-v3": base64.b64encode(digest).decode("ascii"), } HUBSPOT_CONTACT_HYDRATE_PROPERTIES = [ "email", "firstname", "lastname", "company", "country_picklist__sfdc_", "hubspot_owner_id", "jobtitle", "leadsource", "lifecyclestage", "no__of_times_mql", "hs_lead_status", "salesforce_industry", "recent_source", "recent_source_details", "lead_source_details", ] def workflow_action_contact_id(body: dict[str, Any]) -> str: obj = body.get("object") if isinstance(obj, dict): for key in ["objectId", "id"]: value = obj.get(key) if value is not None and str(value).strip(): return str(value).strip() for key in ["objectId", "contactId", "vid"]: value = body.get(key) if value is not None and str(value).strip(): return str(value).strip() return "" def hubspot_contact_record(access_token: str, contact_id: str) -> dict[str, Any]: if not access_token: raise AdapterError(HTTPStatus.SERVICE_UNAVAILABLE, "HubSpot access token is not configured") if not contact_id: raise AdapterError(HTTPStatus.BAD_REQUEST, "workflow action contact objectId is required") query = quote(",".join(HUBSPOT_CONTACT_HYDRATE_PROPERTIES), safe="") request = Request( f"https://api.hubapi.com/crm/v3/objects/contacts/{quote(contact_id, safe='')}?properties={query}&archived=false", method="GET", headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, ) try: with urlopen(request, timeout=20) as response: body = json.loads(response.read().decode("utf-8")) except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") status = ( HTTPStatus.BAD_GATEWAY if retryable_http_status(exc.code) else HTTPStatus.BAD_REQUEST ) raise AdapterError(status, f"failed to hydrate HubSpot contact {contact_id}: HTTP {exc.code}: {detail}") from exc except (URLError, TimeoutError, OSError, json.JSONDecodeError) as exc: raise AdapterError(HTTPStatus.BAD_GATEWAY, f"failed to hydrate HubSpot contact {contact_id}") from exc properties = body.get("properties") if not isinstance(properties, dict): properties = {} record: dict[str, Any] = {"id": str(body.get("id") or contact_id)} for key, value in properties.items(): if value is not None: record[str(key)] = value return record def enrich_workflow_action_body(config: AdapterConfig, body: dict[str, Any]) -> dict[str, Any]: if isinstance(body.get("record"), dict) and body["record"].get("id"): return body contact_id = workflow_action_contact_id(body) record = hubspot_contact_record(config.hubspot_access_token, contact_id) enriched = dict(body) enriched["record"] = record enriched.setdefault("mappingSourceRecord", record) input_fields = dict(enriched.get("inputFields") or {}) for field in HUBSPOT_CONTACT_HYDRATE_PROPERTIES: if input_fields.get(field) in (None, "") and record.get(field) not in (None, ""): input_fields[field] = record[field] enriched["inputFields"] = input_fields return enriched def callback_id_from_workflow_action(body: dict[str, Any]) -> str: value = body.get("callbackId") if isinstance(value, str) and value.strip(): return value.strip() raise AdapterError(HTTPStatus.BAD_REQUEST, "workflow action callbackId is required") class WorkflowActionQueue: def __init__(self, db_path: str): self.db_path = db_path parent = os.path.dirname(os.path.abspath(db_path)) if parent: os.makedirs(parent, exist_ok=True) self.init_db() def connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path, timeout=30) conn.row_factory = sqlite3.Row conn.execute("PRAGMA busy_timeout = 30000") conn.execute("PRAGMA synchronous = FULL") return conn def init_db(self) -> None: with self.connect() as conn: conn.execute("PRAGMA journal_mode = WAL") conn.execute( """ CREATE TABLE IF NOT EXISTS workflow_action_receipts ( callback_id TEXT PRIMARY KEY, body_hash TEXT NOT NULL, raw_body TEXT NOT NULL, request_url TEXT NOT NULL, status TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, response_status INTEGER, response_body TEXT, last_error TEXT, next_attempt_at REAL NOT NULL DEFAULT 0, received_at REAL NOT NULL, updated_at REAL NOT NULL, delivery_policy TEXT NOT NULL DEFAULT 'legacy_replay', final_execution_state TEXT, callback_attempts INTEGER NOT NULL DEFAULT 0, callback_next_attempt_at REAL NOT NULL DEFAULT 0, callback_response_status INTEGER, callback_response_body TEXT, callback_last_error TEXT, crm_receipt_id TEXT, crm_event_id TEXT, result_attempts INTEGER NOT NULL DEFAULT 0, result_next_attempt_at REAL NOT NULL DEFAULT 0, result_last_error TEXT ) """ ) columns = { str(row["name"]) for row in conn.execute("PRAGMA table_info(workflow_action_receipts)").fetchall() } migrations = { "delivery_policy": "TEXT NOT NULL DEFAULT 'legacy_replay'", "final_execution_state": "TEXT", "callback_attempts": "INTEGER NOT NULL DEFAULT 0", "callback_next_attempt_at": "REAL NOT NULL DEFAULT 0", "callback_response_status": "INTEGER", "callback_response_body": "TEXT", "callback_last_error": "TEXT", "crm_receipt_id": "TEXT", "crm_event_id": "TEXT", "result_attempts": "INTEGER NOT NULL DEFAULT 0", "result_next_attempt_at": "REAL NOT NULL DEFAULT 0", "result_last_error": "TEXT", } for name, declaration in migrations.items(): if name not in columns: conn.execute( f"ALTER TABLE workflow_action_receipts ADD COLUMN {name} {declaration}" ) conn.execute("CREATE INDEX IF NOT EXISTS idx_workflow_action_receipts_status_next ON workflow_action_receipts(status, next_attempt_at, received_at)") conn.execute( "CREATE INDEX IF NOT EXISTS idx_workflow_action_receipts_status_callback_next " "ON workflow_action_receipts(status, callback_next_attempt_at, received_at)" ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_workflow_action_receipts_status_result_next " "ON workflow_action_receipts(status, result_next_attempt_at, received_at)" ) self.recover_stale_processing(conn, time.time()) conn.commit() @staticmethod def recover_stale_processing(conn: sqlite3.Connection, now: float) -> None: stale_before = now - WORKFLOW_ACTION_PROCESSING_LEASE_SECONDS conn.execute( "UPDATE workflow_action_receipts SET status='pending', updated_at=? " "WHERE status='processing' AND updated_at <= ?", (now, stale_before), ) conn.execute( "UPDATE workflow_action_receipts SET status='crm_pending', updated_at=? " "WHERE status='crm_processing' AND updated_at <= ?", (now, stale_before), ) conn.execute( "UPDATE workflow_action_receipts SET status='callback_pending', updated_at=? " "WHERE status='callback_processing' AND updated_at <= ?", (now, stale_before), ) conn.execute( "UPDATE workflow_action_receipts SET status='result_pending', updated_at=? " "WHERE status='result_processing' AND updated_at <= ?", (now, stale_before), ) def enqueue( self, callback_id: str, raw_body: bytes, original_url: str, delivery_policy: str = "legacy_replay", ) -> tuple[str, bool]: now = time.time() raw_text = raw_body.decode("utf-8", errors="replace") body_hash = hashlib.sha256(raw_body).hexdigest() initial_status = "crm_pending" if delivery_policy == "must_complete" else "pending" with self.connect() as conn: cursor = conn.execute( """ INSERT OR IGNORE INTO workflow_action_receipts (callback_id, body_hash, raw_body, request_url, status, delivery_policy, next_attempt_at, received_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, 0, ?, ?) """, ( callback_id, body_hash, raw_text, original_url, initial_status, delivery_policy, now, now, ), ) inserted = cursor.rowcount > 0 row = conn.execute( "SELECT status, body_hash, delivery_policy " "FROM workflow_action_receipts WHERE callback_id = ?", (callback_id,), ).fetchone() conn.commit() if row: if str(row["body_hash"]) != body_hash: raise AdapterError( HTTPStatus.CONFLICT, "workflow action callbackId was reused with a different payload", ) if str(row["delivery_policy"]) != delivery_policy: raise AdapterError( HTTPStatus.CONFLICT, "workflow action callbackId was reused with a different delivery policy", ) return str(row["status"]), not inserted raise AdapterError(HTTPStatus.INTERNAL_SERVER_ERROR, "workflow action queue write was not persisted") def claim_next(self) -> dict[str, Any] | None: now = time.time() with self.connect() as conn: conn.execute("BEGIN IMMEDIATE") self.recover_stale_processing(conn, now) row = conn.execute( """ SELECT * FROM workflow_action_receipts WHERE status IN ('pending', 'failed', 'crm_pending', 'crm_failed') AND next_attempt_at <= ? ORDER BY received_at ASC LIMIT 1 """, (now,), ).fetchone() if not row: conn.commit() return None processing_status = ( "crm_processing" if row["delivery_policy"] == "must_complete" else "processing" ) conn.execute( """ UPDATE workflow_action_receipts SET status=?, attempts=attempts+1, updated_at=? WHERE callback_id=? """, (processing_status, now, row["callback_id"]), ) conn.commit() return dict(row) def claim_next_callback(self) -> dict[str, Any] | None: now = time.time() with self.connect() as conn: conn.execute("BEGIN IMMEDIATE") self.recover_stale_processing(conn, now) row = conn.execute( """ SELECT * FROM workflow_action_receipts WHERE status IN ('callback_pending', 'callback_failed') AND callback_next_attempt_at <= ? ORDER BY received_at ASC LIMIT 1 """, (now,), ).fetchone() if not row: conn.commit() return None conn.execute( """ UPDATE workflow_action_receipts SET status='callback_processing', callback_attempts=callback_attempts+1, updated_at=? WHERE callback_id=? """, (now, row["callback_id"]), ) conn.commit() return dict(row) def claim_next_result(self) -> dict[str, Any] | None: now = time.time() with self.connect() as conn: conn.execute("BEGIN IMMEDIATE") self.recover_stale_processing(conn, now) row = conn.execute( """ SELECT * FROM workflow_action_receipts WHERE status IN ('result_pending', 'result_failed') AND result_next_attempt_at <= ? ORDER BY received_at ASC LIMIT 1 """, (now,), ).fetchone() if not row: conn.commit() return None conn.execute( """ UPDATE workflow_action_receipts SET status='result_processing', result_attempts=result_attempts+1, updated_at=? WHERE callback_id=? """, (now, row["callback_id"]), ) conn.commit() return dict(row) def mark_delivered(self, callback_id: str, response_status: int, response_body: str) -> None: now = time.time() with self.connect() as conn: conn.execute( """ UPDATE workflow_action_receipts SET status='delivered', response_status=?, response_body=?, last_error=NULL, updated_at=? WHERE callback_id=? """, (response_status, response_body[:4096], now, callback_id), ) conn.commit() def mark_crm_accepted( self, callback_id: str, response_status: int, response_body: str, *, receipt_id: str = "", event_id: str = "", ) -> None: now = time.time() with self.connect() as conn: conn.execute( """ UPDATE workflow_action_receipts SET status='result_pending', response_status=?, response_body=?, crm_receipt_id=?, crm_event_id=?, result_next_attempt_at=0, last_error=NULL, result_last_error=NULL, updated_at=? WHERE callback_id=? AND delivery_policy='must_complete' """, ( response_status, response_body[:4096], receipt_id or None, event_id or None, now, callback_id, ), ) conn.commit() def mark_result_completed( self, callback_id: str, execution_state: str, *, event_id: str = "", detail: str = "", ) -> None: if execution_state not in {"SUCCESS", "FAIL_CONTINUE"}: raise ValueError(f"unsupported workflow action execution state: {execution_state}") now = time.time() with self.connect() as conn: conn.execute( """ UPDATE workflow_action_receipts SET status='callback_pending', final_execution_state=?, crm_event_id=COALESCE(NULLIF(?, ''), crm_event_id), callback_next_attempt_at=0, result_last_error=CASE WHEN ?='FAIL_CONTINUE' THEN ? ELSE NULL END, updated_at=? WHERE callback_id=? AND delivery_policy='must_complete' """, ( execution_state, event_id, execution_state, detail[:4096] or None, now, callback_id, ), ) conn.commit() def mark_result_waiting( self, callback_id: str, attempts: int, max_attempts: int, message: str, *, event_id: str = "", ) -> None: now = time.time() exhausted = attempts >= max_attempts status = "callback_pending" if exhausted else "result_failed" backoff = min(300, 2 ** max(0, attempts - 1)) next_attempt_at = now + backoff if not exhausted else now with self.connect() as conn: conn.execute( """ UPDATE workflow_action_receipts SET status=?, final_execution_state=CASE WHEN ? THEN 'FAIL_CONTINUE' ELSE final_execution_state END, crm_event_id=COALESCE(NULLIF(?, ''), crm_event_id), result_last_error=?, result_next_attempt_at=?, callback_next_attempt_at=CASE WHEN ? THEN 0 ELSE callback_next_attempt_at END, updated_at=? WHERE callback_id=? AND delivery_policy='must_complete' """, ( status, exhausted, event_id, message[:4096], next_attempt_at, exhausted, now, callback_id, ), ) conn.commit() def mark_callback_completed( self, callback_id: str, response_status: int, response_body: str, ) -> None: now = time.time() with self.connect() as conn: conn.execute( """ UPDATE workflow_action_receipts SET status='completed', callback_response_status=?, callback_response_body=?, callback_last_error=NULL, updated_at=? WHERE callback_id=? """, (response_status, response_body[:4096], now, callback_id), ) conn.commit() def mark_callback_failed( self, callback_id: str, attempts: int, max_attempts: int, message: str, *, response_status: int | None = None, terminal: bool = False, ) -> None: now = time.time() status = "callback_dead" if terminal or attempts >= max_attempts else "callback_failed" backoff = min(300, 2 ** max(0, attempts - 1)) next_attempt_at = now + backoff if status == "callback_failed" else now with self.connect() as conn: conn.execute( """ UPDATE workflow_action_receipts SET status=?, callback_response_status=?, callback_last_error=?, callback_next_attempt_at=?, updated_at=? WHERE callback_id=? """, ( status, response_status, message[:4096], next_attempt_at, now, callback_id, ), ) conn.commit() def mark_failed( self, callback_id: str, attempts: int, max_attempts: int, message: str, *, retryable: bool = True, ) -> None: now = time.time() with self.connect() as conn: row = conn.execute( "SELECT delivery_policy FROM workflow_action_receipts WHERE callback_id=?", (callback_id,), ).fetchone() must_complete = bool(row and row["delivery_policy"] == "must_complete") exhausted = not retryable or attempts >= max_attempts if must_complete and exhausted: status = "callback_pending" elif must_complete: status = "crm_failed" else: status = "dead" if exhausted else "failed" backoff = min(300, 2 ** max(0, attempts - 1)) next_attempt_at = ( now + backoff if status in {"failed", "crm_failed"} else now ) conn.execute( """ UPDATE workflow_action_receipts SET status=?, final_execution_state=CASE WHEN ? THEN 'FAIL_CONTINUE' ELSE final_execution_state END, callback_next_attempt_at=CASE WHEN ? THEN 0 ELSE callback_next_attempt_at END, last_error=?, next_attempt_at=?, updated_at=? WHERE callback_id=? """, ( status, must_complete and exhausted, must_complete and exhausted, message[:4096], next_attempt_at, now, callback_id, ), ) conn.commit() @dataclass(frozen=True) class ConnectorExecutionSnapshot: outcome: str event_id: str = "" detail: str = "" def connector_acceptance_ids(response_body: str) -> tuple[str, str]: try: payload = json.loads(response_body) except json.JSONDecodeError: return "", "" if not isinstance(payload, dict): return "", "" receipt_id = str(payload.get("receiptId") or "").strip() event_id = str(payload.get("eventId") or "").strip() event_ids = payload.get("eventIds") if not event_id and isinstance(event_ids, list): values = [str(value).strip() for value in event_ids if str(value).strip()] if len(values) == 1: event_id = values[0] return receipt_id, event_id def sql_text(value: str) -> str: return f"CONVERT(0x{value.encode('utf-8').hex()} USING utf8mb4)" def connector_result_query(config: AdapterConfig, sql: str) -> str: return common.db_query_scalar( config.runtime_env, sql, database=config.runtime_env["TIDB_DATABASE"], ) def parse_connector_result_json(raw: str, label: str) -> dict[str, Any]: if not raw.strip(): return {} try: value = json.loads(raw) except json.JSONDecodeError as exc: raise RuntimeError(f"invalid {label} result-store response") from exc if not isinstance(value, dict): raise RuntimeError(f"invalid {label} result-store response") return value def connector_execution_snapshot( config: AdapterConfig, *, receipt_id: str, event_id: str, ) -> ConnectorExecutionSnapshot: resolved_event_id = event_id.strip() if receipt_id.strip() and not resolved_event_id: receipt_raw = connector_result_query( config, "SELECT JSON_OBJECT(" "'status', COALESCE(Status, ''), " "'eventIds', COALESCE(EnqueuedEventIds, JSON_ARRAY())) " "FROM `_System_ConnectorIngressReceipt` " f"WHERE Id={sql_text(receipt_id.strip())} " "LIMIT 1", ) receipt = parse_connector_result_json(receipt_raw, "Connector Ingress Receipt") receipt_status = str(receipt.get("status") or "").strip().lower() event_ids = receipt.get("eventIds") if isinstance(event_ids, str): try: event_ids = json.loads(event_ids) except json.JSONDecodeError: event_ids = [] if isinstance(event_ids, list): values = [str(value).strip() for value in event_ids if str(value).strip()] if len(values) == 1: resolved_event_id = values[0] if receipt_status in {"dead", "config_terminal"}: return ConnectorExecutionSnapshot( "failure", detail=f"Connector Ingress Receipt reached terminal state {receipt_status}", ) if not resolved_event_id: return ConnectorExecutionSnapshot( "pending", detail="waiting for Connector Event materialization", ) event_raw = connector_result_query( config, "SELECT JSON_OBJECT(" "'id', Id, " "'status', COALESCE(Status, ''), " "'executionOwner', COALESCE(ExecutionOwner, '')) " "FROM `_System_ConnectorEvent` " f"WHERE Id={sql_text(resolved_event_id)} " "LIMIT 1", ) event = parse_connector_result_json(event_raw, "Connector Event") if not event: return ConnectorExecutionSnapshot( "pending", event_id=resolved_event_id, detail="waiting for Connector Event row", ) status = str(event.get("status") or "").strip().lower() execution_owner = str(event.get("executionOwner") or "").strip().lower() if execution_owner == "work_item": work_raw = connector_result_query( config, "SELECT JSON_OBJECT(" "'count', COUNT(*), " "'state', COALESCE(MAX(State), ''), " "'attempt', COALESCE(MAX(Attempt), 0), " "'failureClass', COALESCE(MAX(FailureClass), '')) " "FROM `_System_ConnectorWorkItem` " f"WHERE EventId={sql_text(resolved_event_id)} " "AND WorkType IN ('connector_event', 'workflow_action_apply')", ) work = parse_connector_result_json(work_raw, "Connector WorkItem") try: work_count = int(work.get("count") or 0) except (TypeError, ValueError) as exc: raise RuntimeError("invalid Connector WorkItem count") from exc if work_count == 0: return ConnectorExecutionSnapshot( "pending", event_id=resolved_event_id, detail="waiting for Connector WorkItem materialization", ) if work_count != 1: return ConnectorExecutionSnapshot( "failure", event_id=resolved_event_id, detail=f"Connector Event has {work_count} execution WorkItems, want 1", ) work_state = str(work.get("state") or "").strip().lower() if work_state == "processed": return ConnectorExecutionSnapshot( "success", event_id=resolved_event_id, detail="Connector WorkItem processed", ) if work_state in {"dead", "skipped"}: failure_class = str(work.get("failureClass") or "").strip() suffix = f" ({failure_class})" if failure_class else "" return ConnectorExecutionSnapshot( "failure", event_id=resolved_event_id, detail=f"Connector WorkItem reached terminal state {work_state}{suffix}", ) return ConnectorExecutionSnapshot( "pending", event_id=resolved_event_id, detail=f"Connector WorkItem is {work_state or 'not ready'}", ) if status == "processed": return ConnectorExecutionSnapshot( "success", event_id=resolved_event_id, detail="Connector Event processed", ) if status in {"dead", "skipped"}: return ConnectorExecutionSnapshot( "failure", event_id=resolved_event_id, detail=f"Connector Event reached terminal state {status}", ) return ConnectorExecutionSnapshot( "pending", event_id=resolved_event_id, detail=f"Connector Event is {status or 'not ready'}", ) class WorkflowActionReplayer(threading.Thread): def __init__(self, config: AdapterConfig, queue: WorkflowActionQueue): super().__init__(daemon=True) self.config = config self.queue = queue self.min_interval = 1.0 / config.workflow_action_replay_rps self.next_phase_index = 0 def claim_next_phase(self) -> tuple[str, dict[str, Any] | None]: phases = ( ("callback", self.queue.claim_next_callback), ("result", self.queue.claim_next_result), ("crm", self.queue.claim_next), ) for offset in range(len(phases)): index = (self.next_phase_index + offset) % len(phases) phase, claim = phases[index] row = claim() if row: self.next_phase_index = (index + 1) % len(phases) return phase, row return "", None def run(self) -> None: while True: phase, row = self.claim_next_phase() if not row: time.sleep(0.25) continue started = time.monotonic() if phase == "callback": self.complete_callback(row) elif phase == "result": self.observe_result(row) else: self.replay(row) elapsed = time.monotonic() - started if elapsed < self.min_interval: time.sleep(self.min_interval - elapsed) def replay(self, row: dict[str, Any]) -> None: callback_id = str(row["callback_id"]) attempts = int(row["attempts"]) + 1 raw_body = str(row["raw_body"]).encode("utf-8") try: body = parse_json_body(raw_body) replay_body = enrich_workflow_action_body(self.config, body) replay_raw_body = json.dumps(replay_body, separators=(",", ":")).encode("utf-8") headers = { "Content-Type": "application/json", **hubspot_signature_headers( self.config.workflow_action_replay_signing_secret, "POST", self.config.workflow_action_replay_signing_url, replay_raw_body, ), } request = Request( self.config.workflow_action_replay_url, data=replay_raw_body, method="POST", headers=headers, ) with urlopen(request, timeout=20) as response: response_body = response.read().decode("utf-8", errors="replace") status = int(response.status) if 200 <= status < 300: if row["delivery_policy"] == "must_complete": receipt_id, event_id = connector_acceptance_ids(response_body) if not receipt_id and not event_id: self.queue.mark_failed( callback_id, attempts, self.config.workflow_action_max_attempts, "NexusCRM accepted the workflow action without a durable receipt or event id", ) return self.queue.mark_crm_accepted( callback_id, status, response_body, receipt_id=receipt_id, event_id=event_id, ) else: self.queue.mark_delivered(callback_id, status, response_body) return self.queue.mark_failed( callback_id, attempts, self.config.workflow_action_max_attempts, f"NexusCRM returned HTTP {status}: {response_body}", retryable=retryable_http_status(status), ) except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") self.queue.mark_failed( callback_id, attempts, self.config.workflow_action_max_attempts, f"NexusCRM returned HTTP {exc.code}: {body}", retryable=retryable_http_status(exc.code), ) except AdapterError as exc: self.queue.mark_failed( callback_id, attempts, self.config.workflow_action_max_attempts, exc.message, retryable=retryable_http_status(exc.status), ) except (URLError, TimeoutError, OSError) as exc: self.queue.mark_failed(callback_id, attempts, self.config.workflow_action_max_attempts, f"NexusCRM replay failed: {exc}") def observe_result(self, row: dict[str, Any]) -> None: callback_id = str(row["callback_id"]) attempts = int(row["result_attempts"]) + 1 try: snapshot = connector_execution_snapshot( self.config, receipt_id=str(row["crm_receipt_id"] or ""), event_id=str(row["crm_event_id"] or ""), ) if snapshot.outcome == "success": self.queue.mark_result_completed( callback_id, "SUCCESS", event_id=snapshot.event_id, detail=snapshot.detail, ) return if snapshot.outcome == "failure": self.queue.mark_result_completed( callback_id, "FAIL_CONTINUE", event_id=snapshot.event_id, detail=snapshot.detail, ) return self.queue.mark_result_waiting( callback_id, attempts, self.config.workflow_action_result_max_attempts, snapshot.detail, event_id=snapshot.event_id, ) except Exception as exc: self.queue.mark_result_waiting( callback_id, attempts, self.config.workflow_action_result_max_attempts, f"Connector Event result readback failed: {exc}", ) def complete_callback(self, row: dict[str, Any]) -> None: callback_id = str(row["callback_id"]) attempts = int(row["callback_attempts"]) + 1 try: final_state = str(row["final_execution_state"] or "") if final_state not in {"SUCCESS", "FAIL_CONTINUE"}: raise AdapterError( HTTPStatus.INTERNAL_SERVER_ERROR, "workflow action final execution state is missing", ) callback_body = { "outputFields": {"hs_execution_state": final_state}, "typedOutputs": {}, } raw_callback_body = json.dumps( callback_body, separators=(",", ":") ).encode("utf-8") request = Request( f"{WORKFLOW_ACTION_CALLBACK_BASE_URL}/{quote(callback_id, safe='')}/complete", data=raw_callback_body, method="POST", headers={ "Authorization": f"Bearer {self.config.workflow_action_callback_access_token}", "Content-Type": "application/json", }, ) with urlopen(request, timeout=20) as response: response_body = response.read().decode("utf-8", errors="replace") status = int(response.status) if 200 <= status < 300: self.queue.mark_callback_completed( callback_id, status, response_body ) return self.queue.mark_callback_failed( callback_id, attempts, self.config.workflow_action_callback_max_attempts, f"HubSpot callback returned HTTP {status}: {response_body}", response_status=status, terminal=status < 500 and status != 429, ) except HTTPError as exc: response_body = exc.read().decode("utf-8", errors="replace") if exc.code == HTTPStatus.CONFLICT: self.queue.mark_callback_completed( callback_id, exc.code, response_body ) return self.queue.mark_callback_failed( callback_id, attempts, self.config.workflow_action_callback_max_attempts, f"HubSpot callback returned HTTP {exc.code}: {response_body}", response_status=exc.code, terminal=exc.code < 500 and exc.code != HTTPStatus.TOO_MANY_REQUESTS, ) except AdapterError as exc: self.queue.mark_callback_failed( callback_id, attempts, self.config.workflow_action_callback_max_attempts, exc.message, terminal=True, ) except (URLError, TimeoutError, OSError) as exc: self.queue.mark_callback_failed( callback_id, attempts, self.config.workflow_action_callback_max_attempts, f"HubSpot callback failed: {exc}", ) class OptionsHandler(BaseHTTPRequestHandler): server: "OptionsServer" def do_GET(self) -> None: self.handle_options_request({}) def do_POST(self) -> None: parsed = urlparse(self.path) if parsed.path == WORKFLOW_ACTION_EXECUTE_PATH: self.handle_workflow_action_execute(parsed) return try: body = read_json_body(self) except AdapterError as exc: json_response(self, exc.status, {"message": exc.message}) return self.handle_options_request(body) def handle_options_request(self, body: dict[str, Any]) -> None: parsed = urlparse(self.path) query = parse_qs(parsed.query) try: if parsed.path == "/hubspot/workflow-actions/options/health": json_response(self, HTTPStatus.OK, {"ok": True}) return if parsed.path == "/hubspot/workflow-actions/health": json_response(self, HTTPStatus.OK, {"ok": True}) return verify_options_token(self.server.config, query, body) if parsed.path == "/hubspot/workflow-actions/options/campaigns": json_response(self, HTTPStatus.OK, campaign_options(self.server.config, query, body)) return if parsed.path == "/hubspot/workflow-actions/options/statuses": json_response(self, HTTPStatus.OK, status_options(self.server.config, query, body)) return raise AdapterError(HTTPStatus.NOT_FOUND, "unknown options path") except AdapterError as exc: json_response(self, exc.status, {"message": exc.message}) except Exception: json_response(self, HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "internal adapter error"}) def handle_workflow_action_execute(self, parsed: Any) -> None: try: if self.server.config.workflow_action_execute_mode != "proxy": raise AdapterError(HTTPStatus.NOT_FOUND, "workflow action execution is blocked") raw = read_raw_body(self) original_url = request_url(self) verify_hubspot_v3_signature(self.server.config, "POST", original_url, raw, self.headers) body = parse_json_body(raw) policy = workflow_action_policy(self.server.config, body) if policy in {"legacy_replay", "must_complete"}: callback_id = callback_id_from_workflow_action(body) status, duplicate = self.server.workflow_action_queue.enqueue( callback_id, raw, original_url, policy ) if policy == "must_complete": json_response( self, HTTPStatus.OK, { "outputFields": { "hs_execution_state": "BLOCK", "hs_expiration_duration": "P1WT1H", } }, ) return json_response( self, HTTPStatus.ACCEPTED, { "accepted": True, "callbackId": callback_id, "duplicate": duplicate, "status": status, }, ) return json_response( self, HTTPStatus.OK, {"outputFields": {"hs_execution_state": "FAIL_CONTINUE"}}, ) except AdapterError as exc: json_response(self, exc.status, {"message": exc.message}) except Exception: json_response(self, HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "internal adapter error"}) def log_message(self, fmt: str, *args: Any) -> None: message = fmt % tuple(unquote(str(arg)).replace(self.server.config.options_token, "REDACTED") for arg in args) print(f"{self.address_string()} - {message}", flush=True) class OptionsServer(ThreadingHTTPServer): allow_reuse_address = True daemon_threads = True def __init__(self, server_address: tuple[str, int], config: AdapterConfig): self.config = config self.request_queue_size = config.workflow_action_adapter_backlog super().__init__(server_address, OptionsHandler, bind_and_activate=False) try: self.server_bind() self.server_activate() except Exception: self.server_close() raise self.workflow_action_queue = WorkflowActionQueue(config.workflow_action_queue_db) if ( config.workflow_action_execute_mode == "proxy" and {"legacy_replay", "must_complete"} & set(config.workflow_action_policies.values()) ): WorkflowActionReplayer(config, self.workflow_action_queue).start() def main() -> int: config = load_config() server = OptionsServer((config.listen_host, config.listen_port), config) print( f"HubSpot workflow adapter listening on {config.listen_host}:{config.listen_port}", flush=True, ) server.serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())