from __future__ import annotations import hashlib import json from collections.abc import Iterable, Mapping from dataclasses import dataclass from datetime import UTC, datetime from decimal import Decimal, InvalidOperation from typing import Any, Protocol from .netsuite_revenue_admission import AccountLookup, AdmittedRow from .netsuite_revenue_candidate import CandidateStore from .netsuite_revenue_config import RevenueSyncConfig from .nexuscrm_client import UpsertResult class NexusCRMRevenueContractError(RuntimeError): pass class NexusCRMRevenueClient(Protocol): def describe_object(self, object_name: str) -> dict[str, Any]: ... def resolve_active_account_matches(self, external_ids: list[str]) -> dict[str, list[str]]: ... def query_records(self, soql: str) -> list[dict[str, Any]]: ... def upsert_collection( self, *, object_name: str, external_id_field: str, records: list[dict[str, Any]] ) -> list[UpsertResult]: ... @dataclass(frozen=True) class APIWritePlan: records: tuple[dict[str, Any], ...] created_or_changed: int last_seen_only: int @dataclass(frozen=True) class APIWriteResult: attempted: int created: int updated: int failed: int class CRMAPIAccountLookup(AccountLookup): def __init__(self, client: NexusCRMRevenueClient) -> None: self.client = client def find_active_accounts(self, account_ids: list[str]) -> Mapping[str, list[str]]: return self.client.resolve_active_account_matches(account_ids) def describe_sha256(describe: Mapping[str, Any]) -> str: return hashlib.sha256( json.dumps(describe, sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str).encode() ).hexdigest() def validate_target_contract(config: RevenueSyncConfig, describe: Mapping[str, Any]) -> None: """Fail closed unless the live Describe response satisfies Design v0.8.""" if describe.get("name") not in {None, config.target["object"]}: raise NexusCRMRevenueContractError("CRM Describe object identity drift") fields = _describe_fields(describe) target = config.target required = { target["name_field"], *(mapping.field for mapping in config.mappings), target["external_key_field"], target["account"]["relation_field"], *target["controls"].values(), } missing = sorted(required - fields.keys()) if missing: raise NexusCRMRevenueContractError("CRM object is missing fields: " + ", ".join(missing)) required_on_admission = { target["name_field"], target["external_key_field"], config.account_id_mapping.field, target["account"]["relation_field"], *( value for name, value in target["controls"].items() if name != "accounting_book_id" ), } optional = sorted(name for name in required_on_admission if fields[name].get("nillable") is not False) if optional: raise NexusCRMRevenueContractError("CRM fields must be metadata-required: " + ", ".join(optional)) external = fields[target["external_key_field"]] if external.get("unique") is not True: raise NexusCRMRevenueContractError("NetSuite_Source_Key__c must be unique") not_createable = sorted(name for name in required if fields[name].get("createable") is not True) if not_createable: raise NexusCRMRevenueContractError("integration user lacks create access: " + ", ".join(not_createable)) updatable = { target["name_field"], *(mapping.field for mapping in config.mappings), target["external_key_field"], target["account"]["relation_field"], *( value for name, value in target["controls"].items() if name not in {"first_seen_run_id", "first_seen_at", "is_source_active"} ), } not_updateable = sorted(name for name in updatable if fields[name].get("updateable") is not True) if not_updateable: raise NexusCRMRevenueContractError("integration user lacks update access: " + ", ".join(not_updateable)) _validate_field_types(config, fields) def read_existing_records( config: RevenueSyncConfig, client: NexusCRMRevenueClient, source_keys: Iterable[str] ) -> dict[str, dict[str, Any]]: target = config.target fields = ( target["external_key_field"], target["controls"]["payload_hash"], target["account"]["relation_field"], target["active_flag_field"], *(_amount_fields(config)), ) by_key: dict[str, dict[str, Any]] = {} for batch in _batches(sorted(set(source_keys)), 100): literals = ", ".join(f"'{_soql_string(value)}'" for value in batch) soql = ( f"SELECT {', '.join(fields)} FROM {target['object']} " f"WHERE {target['external_key_field']} IN ({literals})" ) for record in client.query_records(soql): source_key = str(record.get(target["external_key_field"]) or "").strip() if not source_key or source_key in by_key: raise NexusCRMRevenueContractError("CRM target query returned blank or duplicate source keys") by_key[source_key] = record return by_key def build_write_plan( config: RevenueSyncConfig, admitted: Iterable[AdmittedRow], *, existing_by_source_key: Mapping[str, Mapping[str, Any]], run_id: str, posting_period_id: str, seen_at: str | None = None, ) -> APIWritePlan: """Create explicit Records API payloads without CRM-owned system fields.""" seen_at = seen_at or datetime.now(UTC).isoformat() records: list[dict[str, Any]] = [] created_or_changed = 0 last_seen_only = 0 source_keys: set[str] = set() for admitted_row in admitted: candidate = admitted_row.candidate if candidate.source_key in source_keys: raise NexusCRMRevenueContractError("admitted candidate repeats a NetSuite source key") source_keys.add(candidate.source_key) if not admitted_row.crm_account_id or candidate.account_id is None: raise NexusCRMRevenueContractError("admitted record lacks required Account association") if candidate.payload.get(config.account_id_mapping.field) != candidate.account_id: raise NexusCRMRevenueContractError("source Account raw/display evidence drift") existing = existing_by_source_key.get(candidate.source_key) if existing is not None and not _is_active(existing.get(config.target["active_flag_field"])): raise NexusCRMRevenueContractError("candidate conflicts with an inactive target source key") relation_field = config.target["account"]["relation_field"] if existing is not None and str(existing.get(relation_field) or "") != admitted_row.crm_account_id: raise NexusCRMRevenueContractError("existing CRM_Account__c differs from exact Account admission") controls = _mutable_controls( config, admitted_row, run_id=run_id, posting_period_id=posting_period_id, seen_at=seen_at, ) payload: dict[str, Any] = { "attributes": {"type": config.target["object"]}, config.target["external_key_field"]: candidate.source_key, **controls, } hash_field = config.target["controls"]["payload_hash"] if existing is not None and str(existing.get(hash_field) or "") == candidate.payload_hash: last_seen_only += 1 else: payload.update(candidate.payload) payload[config.target["name_field"]] = _record_name(config, admitted_row) payload[config.target["account"]["relation_field"]] = admitted_row.crm_account_id if existing is None: payload.update(_first_seen_controls(config, run_id=run_id, seen_at=seen_at)) payload[config.target["controls"]["is_source_active"]] = True created_or_changed += 1 _assert_payload_whitelist(config, payload) records.append(payload) return APIWritePlan(records=tuple(records), created_or_changed=created_or_changed, last_seen_only=last_seen_only) def write_records( config: RevenueSyncConfig, client: NexusCRMRevenueClient, store: CandidateStore, plan: APIWritePlan, *, attempt_prefix: str, ) -> APIWriteResult: attempts: list[dict[str, Any]] = [] result = APIWriteResult(attempted=0, created=0, updated=0, failed=0) for batch in _batches(list(plan.records), int(config.execution["crm_upsert_batch_size"])): batch_results = client.upsert_collection( object_name=config.target["object"], external_id_field=config.target["external_key_field"], records=batch, ) if len(batch_results) != len(batch): raise NexusCRMRevenueContractError("CRM composite result count drift") for payload, item in zip(batch, batch_results, strict=True): attempts.append( { "source_key": payload[config.target["external_key_field"]], "payload_hash": payload.get(config.target["controls"]["payload_hash"]), "success": item.success, "created": item.created, "error_count": len(item.errors), } ) result = APIWriteResult( attempted=result.attempted + 1, created=result.created + int(item.success and item.created is True), updated=result.updated + int(item.success and item.created is not True), failed=result.failed + int(not item.success), ) if result.failed: break store.write_jsonl(f"{attempt_prefix}/records-api-attempt-ledger.jsonl", attempts) if result.failed: raise NexusCRMRevenueContractError(f"CRM composite upsert rejected {result.failed} admitted record(s)") return result def verify_active_target_parity( config: RevenueSyncConfig, *, admitted: Iterable[AdmittedRow], existing_by_source_key: Mapping[str, Mapping[str, Any]], ) -> dict[str, Any]: admitted_rows = list(admitted) expected_keys = {row.candidate.source_key for row in admitted_rows} active = { key: value for key, value in existing_by_source_key.items() if _is_active(value.get(config.target["active_flag_field"])) } if set(active) != expected_keys: raise NexusCRMRevenueContractError("target admitted active source-key parity failed") for admitted_row in admitted_rows: record = active[admitted_row.candidate.source_key] if str(record.get(config.target["controls"]["payload_hash"]) or "") != admitted_row.candidate.payload_hash: raise NexusCRMRevenueContractError("target payload-hash parity failed") if str(record.get(config.target["account"]["relation_field"]) or "") != admitted_row.crm_account_id: raise NexusCRMRevenueContractError("target CRM_Account__c parity failed") expected_totals = _amount_totals(config, (row.candidate.payload for row in admitted_rows)) actual_totals = _amount_totals(config, active.values()) if expected_totals != actual_totals: raise NexusCRMRevenueContractError("target amount parity failed") return { "target_active_count": len(active), "target_active_hash": _target_hash(config, active), "target_amount_totals": actual_totals, } def _describe_fields(describe: Mapping[str, Any]) -> dict[str, Mapping[str, Any]]: raw = describe.get("fields") if not isinstance(raw, list): raise NexusCRMRevenueContractError("CRM Describe lacks fields") fields: dict[str, Mapping[str, Any]] = {} for field in raw: if not isinstance(field, Mapping): raise NexusCRMRevenueContractError("CRM Describe field is invalid") name = str(field.get("name") or field.get("apiName") or "") if not name or name in fields: raise NexusCRMRevenueContractError("CRM Describe has blank/duplicate field names") fields[name] = field return fields def _validate_field_types(config: RevenueSyncConfig, fields: Mapping[str, Mapping[str, Any]]) -> None: expected: dict[str, set[str]] = { config.target["name_field"]: {"string", "text"}, config.target["external_key_field"]: {"string", "text"}, config.account_id_mapping.field: {"string", "text"}, config.target["account"]["relation_field"]: {"reference"}, config.target["controls"]["first_seen_at"]: {"datetime"}, config.target["controls"]["last_seen_at"]: {"datetime"}, config.target["controls"]["is_source_active"]: {"boolean"}, } for mapping in config.mappings: expected[mapping.field] = { "text": {"string", "text", "textarea"}, "currency": {"currency"}, "number": {"double"}, "date": {"date"}, "datetime": {"datetime"}, }[mapping.kind] mismatched = sorted( name for name, accepted in expected.items() if str(fields[name].get("type") or "").lower() not in accepted ) if mismatched: raise NexusCRMRevenueContractError("CRM field type contract drift: " + ", ".join(mismatched)) def _mutable_controls( config: RevenueSyncConfig, admitted: AdmittedRow, *, run_id: str, posting_period_id: str, seen_at: str, ) -> dict[str, Any]: controls = config.target["controls"] identity = admitted.candidate.source_identity return { controls["account_realm"]: config.source["account_realm"], controls["transaction_id"]: identity.get("transactionInternalId"), controls["line_unique_key"]: identity.get("lineUniqueKey"), controls["accounting_book_id"]: identity.get("accountingBookId"), controls["posting_period_id"]: posting_period_id, controls["search_id"]: config.source["saved_search_id"], controls["schema_version"]: config.source["report_schema_version"], controls["payload_hash"]: admitted.candidate.payload_hash, controls["last_seen_run_id"]: run_id, controls["last_seen_at"]: seen_at, } def _first_seen_controls(config: RevenueSyncConfig, *, run_id: str, seen_at: str) -> dict[str, Any]: controls = config.target["controls"] return {controls["first_seen_run_id"]: run_id, controls["first_seen_at"]: seen_at} def _assert_payload_whitelist(config: RevenueSyncConfig, payload: Mapping[str, Any]) -> None: allowed = { "attributes", config.target["name_field"], *(mapping.field for mapping in config.mappings), config.target["external_key_field"], config.target["account"]["relation_field"], *config.target["controls"].values(), } unexpected = sorted(set(payload) - allowed) protected = sorted(set(payload) & set(config.target["protected_system_fields"])) if unexpected or protected: details = [*(f"unexpected={name}" for name in unexpected), *(f"protected={name}" for name in protected)] raise NexusCRMRevenueContractError("Records API payload whitelist drift: " + ", ".join(details)) if any(mapping.field in payload for mapping in config.mappings): relation = config.target["account"]["relation_field"] if not payload.get(relation): raise NexusCRMRevenueContractError("admitted write payload lacks CRM_Account__c") def _amount_fields(config: RevenueSyncConfig) -> tuple[str, str]: return ( next(mapping.field for mapping in config.mappings if mapping.header == "Transaction Amount"), next(mapping.field for mapping in config.mappings if mapping.header == "Amount"), ) def _amount_totals(config: RevenueSyncConfig, rows: Iterable[Mapping[str, Any]]) -> dict[str, str]: fields = _amount_fields(config) totals = {field: Decimal("0") for field in fields} for row in rows: for field in fields: try: value = Decimal(str(row.get(field))) except (InvalidOperation, ValueError) as error: raise NexusCRMRevenueContractError(f"amount field {field} is not numeric") from error if not value.is_finite(): raise NexusCRMRevenueContractError(f"amount field {field} is not finite") totals[field] += value return {field: str(value.normalize()) for field, value in totals.items()} def _target_hash(config: RevenueSyncConfig, active: Mapping[str, Mapping[str, Any]]) -> str: digest = hashlib.sha256() for source_key, row in sorted(active.items()): digest.update(source_key.encode()) digest.update(b"\0") digest.update(str(row.get(config.target["controls"]["payload_hash"]) or "").encode()) digest.update(b"\0") digest.update(str(row.get(config.target["account"]["relation_field"]) or "").encode()) digest.update(b"\n") return digest.hexdigest() def _record_name(config: RevenueSyncConfig, admitted: AdmittedRow) -> str: document_field = next(mapping.field for mapping in config.mappings if mapping.header == "Document Number") document = str(admitted.candidate.payload.get(document_field) or "").strip() line = str(admitted.candidate.source_identity.get("lineUniqueKey") or "").strip() return f"{document or admitted.candidate.source_key} / {line or admitted.candidate.source_key}" def _is_active(value: Any) -> bool: return value is True or str(value).strip().lower() in {"1", "true", "yes"} def _soql_string(value: str) -> str: return value.replace("\\", "\\\\").replace("'", "\\'") def _batches(values: list[Any], size: int) -> list[list[Any]]: return [values[start : start + size] for start in range(0, len(values), size)]