#!/usr/bin/env python3
# Runtime: stdlib-only (json, urllib). No third-party deps.
# Kept 3.9-compatible on purpose so it can run under the macOS launchd
# interpreter (/usr/bin/python3 == 3.9.6) as well as CI runners:
#   - NO nested f-strings that reuse the same quote char
#   - NO match/case or other 3.10+ syntax
#   - build dynamic SOQL by string concatenation, not nested f-strings
# Validate before deploy:
#   python3 -c "import ast,sys; ast.parse(open(sys.argv[1]).read())" <file>
"""
NexusCRM Prod - Org auto-create reconciliation + backfill.

Finds Leads with Portal_ID__c populated but no corresponding Org__c and
backfills by re-triggering Lead_Auto_Create_Org. Root cause: ETL/sync writes
Leads directly (bypassing the after-save trigger pipeline), so the Org-create
flow never fires at load. This job is the safety net.

Re-trigger mechanism: Lead_Auto_Create_Org is RecordAfterSave/CreateAndUpdate
with entry filter `Portal_ID__c IsNull = false` and does NOT require the record
to change - so ANY save while Portal_ID__c is non-null re-fires it. We therefore
do a single write to the ops-only field Reconcile_Touch__c (datetime = now).
Get_Org dedups, so re-firing is idempotent.

SIDE EFFECTS - read this before changing anything here.

This is a Lead SAVE, so it re-runs every Lead flow whose entry condition holds -
not just the Org flow. Note that this is not special to this job: ANY Lead save
from any source (a rep editing a field in the UI, an app-layer ETL write, a
HubSpot sync) re-runs exactly the same set. Audited against the merged pack
(12 Lead-triggered flows), what our save actually does:

  no-ops (fire, but change nothing for us):
    - lead_convert_opportunity  (before-save, unfiltered) derives
      Need_Create_Opportunity__c purely from Opportunity_Name__c, so re-running
      it on unchanged state recomputes the value it already has.
    - BN_Lead_Relate_to_Account_by_Email_Domain_Flow (after-save, unfiltered)
      immediately gates on "Email changed OR BN_Account__c changed OR is new".
      We change neither, so it does nothing.

  fill-if-null defaulting (can only populate a NULL field, never overwrite):
    - LeadQuicklySet        Account_Status__c IsNull + Portal_ID not null
                            -> Account_Status__c = "new_user"  (our population)
    - Lead_Quick_Set_Region Region__c IsNull + CountryCode not null -> Region__c
    - Country_is_empty      IP_Country__c = japan + Country IsNull -> Country

  do not fire (gated on fields this job never writes; targets are non-converted):
    Lead_SetCountryCodeFromCountry, Lead_Status_Duration, IQM_Completion_Flow,
    Lead_Assignment_Rule_modification_for_NA_EMEA, Lead_Convert_Filed_Set,
    Lifecycle_Stage_to_Opportunity_Upon_Conversion.

Net effect: no existing business value can be overwritten. The only real
consequence is that platform-intended defaults get filled sooner than the next
save from any other source would have filled them. This is an accepted,
owner-reviewed trade-off.

The alternative with no Lead save at all is for this job to create Org__c
directly instead of re-triggering the flow; it was considered and rejected so
the flow stays the single authority for Org creation.

Still, this is strictly less exposure than the Portal_ID null->value toggle it
replaces, which saved the Lead TWICE (firing all of the above twice, once with
Portal_ID = null), mutated a business field, and could permanently lose
Portal_ID if the process died between its two writes.

Residual duplicate risk: Org creation is asynchronous and Org__c.ID__c is not
unique, so the dedup guard is point-in-time only - a restart, or another save of
the same Lead before the flow settles, can still race into a duplicate Org. The
job cannot close this from the outside; it detects duplicates among the Leads it
touched, reports them, and fails the run.

Credentials (CI path only):
  - Requires PROD_URL, PROD_USERNAME, PROD_PASSWORD env vars. There is no local
    file / Keychain fallback: the script fails closed if any is missing.

Data-in-logs policy:
  - By default only aggregate COUNTS are logged. Record identifiers (Lead Ids,
    tenant Portal Ids) are customer data and must not land in persistent CI
    logs/artifacts, so they are emitted only when LOG_RECORD_IDS=1 (local ops
    use), never in CI.

Logging:
  - Always prints to stdout (captured by CI). Additionally appends to a file
    when RECONCILE_LOG is set.

Exit code:
  - Non-zero if any backfill PATCH hard-fails (so a CI run goes red instead of
    reporting a false-green success). Transient async lag (still_missing_after)
    is NOT treated as failure.
"""
import json, os, re, urllib.request, urllib.parse, urllib.error, time, sys
from pathlib import Path
from datetime import datetime, timezone, timedelta

LOOKBACK_HOURS = 72       # reconcile leads modified in the last N hours (rolling window)
MAX_PER_RUN    = 600      # safety cap on backfills per run
RECONCILE_FIELD = "Reconcile_Touch__c"   # ops-only trigger field (L2 pack)
# `nexuscrm-ui` is the browser client: the target admits it on the password grant
# only for requests carrying browser session-control evidence, which this
# server-to-server caller cannot produce. Bind an installed programmatic
# Connected App through the environment instead.
BROWSER_SESSION_CLIENT_ID = "nexuscrm-ui"
OAUTH_CLIENT_ID     = os.environ.get("NEXUSCRM_OAUTH_CLIENT_ID", "")
OAUTH_CLIENT_SECRET = os.environ.get("NEXUSCRM_OAUTH_CLIENT_SECRET", "")
PORTAL_RE = re.compile(r"^[0-9]{1,32}$")   # Portal IDs are numeric; anything else is anomalous
LOG_IDS = os.environ.get("LOG_RECORD_IDS") == "1"

# --- logging (stdout always; file when RECONCILE_LOG is set) -------------------
LOG = Path(os.environ["RECONCILE_LOG"]) if os.environ.get("RECONCILE_LOG") else None

def log(msg):
    line = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + "  " + msg
    print(line, flush=True)
    if LOG is not None:
        with LOG.open("a") as f:
            f.write(line + "\n")

def soql_lit(v):
    """Escape a value for safe inclusion inside a single-quoted SOQL literal."""
    return str(v).replace("\\", "\\\\").replace("'", "\\'")

# --- credentials --------------------------------------------------------------
def load_creds():
    url = os.environ.get("PROD_URL")
    user = os.environ.get("PROD_USERNAME")
    pwd = os.environ.get("PROD_PASSWORD")
    missing = [n for n, v in (
        ("PROD_URL", url),
        ("PROD_USERNAME", user),
        ("PROD_PASSWORD", pwd),
        ("NEXUSCRM_OAUTH_CLIENT_ID", OAUTH_CLIENT_ID),
        ("NEXUSCRM_OAUTH_CLIENT_SECRET", OAUTH_CLIENT_SECRET),
    ) if not v]
    if missing:
        raise RuntimeError("missing required env var(s): " + ", ".join(missing))
    if OAUTH_CLIENT_ID == BROWSER_SESSION_CLIENT_ID:
        raise RuntimeError(
            f"NEXUSCRM_OAUTH_CLIENT_ID {BROWSER_SESSION_CLIENT_ID!r} is the browser client "
            "and is refused by browser session-control admission; bind an installed "
            "programmatic Connected App instead"
        )
    return url, user, pwd

BASE, USER, PWD = load_creds()

# 408 and 429 are the 4xx codes the target uses for "try again later" -- 429 in
# particular is how the hourly API quota is signalled, and it clears on its own.
# Every other 4xx on the token endpoint is a deterministic credential or client
# rejection that eight retries cannot turn into a token.
RETRYABLE_HTTP_STATUSES = (408, 429)

def token_status_is_retryable(code):
    """Whether a token-endpoint HTTP status is worth another attempt."""
    return not (400 <= code < 500) or code in RETRYABLE_HTTP_STATUSES

def token():
    d = urllib.parse.urlencode({
        "grant_type": "password",
        "client_id": OAUTH_CLIENT_ID,
        "client_secret": OAUTH_CLIENT_SECRET,
        "username": USER,
        "password": PWD,
    }).encode()
    last = "no attempt was made"
    for attempt in range(8):
        try:
            return json.loads(urllib.request.urlopen(
                urllib.request.Request(BASE + "/services/oauth2/token", data=d), timeout=25
            ).read())["access_token"]
        except urllib.error.HTTPError as e:
            # The body carries only OAuth error/error_description, never a secret.
            last = f"HTTP {e.code} {e.read().decode('utf-8', 'replace')[:300]}"
            # A deterministic 4xx buys nothing from eight retries and buries the
            # reason under three minutes of sleeping. Rate limiting still retries.
            if not token_status_is_retryable(e.code):
                break
        except Exception as e:
            last = f"{type(e).__name__}: {e}"
        if attempt < 7:
            time.sleep(6)
    raise RuntimeError(f"token failed: {last}")
TOK = token()

def H():
    return {"Authorization": "Bearer " + TOK, "Content-Type": "application/json"}

def soql_all(q, t=90):
    recs = []
    url = BASE + "/services/data/v63.0/query?q=" + urllib.parse.quote(q)
    while url:
        for _ in range(8):
            try:
                dd = json.loads(urllib.request.urlopen(urllib.request.Request(url, headers=H()), timeout=t).read())
                break
            except Exception:
                time.sleep(6)
        else:
            raise RuntimeError("query failed")
        recs += dd["records"]
        nxt = dd.get("nextRecordsUrl")
        url = (BASE + nxt) if nxt else None
    return recs

def patch_once(lid, body, t=70):
    """Exactly one PATCH attempt. Returns (kind, status, body_text) where kind is
    one of:
      ok         - committed (2xx)
      retry_safe - rejected before processing (401); definitely NOT committed
      ambiguous  - 5xx / timeout / transport error; the save MAY have committed
      fatal      - other 4xx; definitely not committed, do not retry
    The caller must never blindly retry an `ambiguous` result: a second save
    would re-fire the async Org flow, and because Org__c.ID__c is not unique
    that can produce duplicate Org records."""
    global TOK
    try:
        st = urllib.request.urlopen(urllib.request.Request(
            BASE + "/services/data/v63.0/sobjects/Lead/" + urllib.parse.quote(lid),
            data=json.dumps(body).encode(), headers=H(), method="PATCH"), timeout=t).status
        return "ok", st, ""
    except urllib.error.HTTPError as e:
        try:
            eb = e.read().decode("utf-8", "replace")
        except Exception:
            eb = ""
        if e.code == 401:
            TOK = token()
            return "retry_safe", "ERR401", eb
        if e.code in (502, 503, 504):
            return "ambiguous", "ERR" + str(e.code), eb
        return "fatal", "ERR" + str(e.code), eb
    except Exception as e:
        return "ambiguous", "ERR", type(e).__name__

def locked_body(body):
    """True if an error body indicates the Lead is locked (converted)."""
    return ("ENTITY_IS_LOCKED" in body or "IsConverted" in body
            or "read-only" in body.lower())

def trigger_touch(lid, prev_value):
    """Write the ops trigger field with at-most-once semantics.

    Returns ('done'|'skip'|'error', detail). After an ambiguous failure we do NOT
    blindly resubmit: we re-read the field and retry only when the write
    demonstrably did not land, so the Lead is saved at most once and the async
    Org flow cannot be double-fired into creating duplicate Orgs."""
    lit = soql_lit(lid)
    for _ in range(3):
        stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000+0000")
        kind, st, body = patch_once(lid, {RECONCILE_FIELD: stamp})
        if kind == "ok":
            return "done", ""
        if kind == "fatal":
            # A Lead that converted between scan and write is locked - benign race.
            return ("skip" if locked_body(body) else "error"), str(st) + " " + body
        if kind == "ambiguous":
            cur = soql_all("SELECT " + RECONCILE_FIELD + " FROM Lead WHERE Id='" + lit + "'")
            if cur and cur[0].get(RECONCILE_FIELD) != prev_value:
                return "done", ""          # it did land despite the ambiguous response
            time.sleep(5)                  # definitively not landed -> safe to retry
        # retry_safe (401): token refreshed, nothing committed -> loop
    return "error", "exhausted"

def field_exists(field):
    """True if `field` is present on the Lead object in the target org."""
    d = json.loads(urllib.request.urlopen(urllib.request.Request(
        BASE + "/services/data/v63.0/sobjects/Lead/describe", headers=H()), timeout=90).read())
    return any(f.get("name") == field for f in d.get("fields", []))

def batched(l, n=170):
    for i in range(0, len(l), n):
        yield l[i:i + n]

def find_missing():
    since = datetime.now(timezone.utc) - timedelta(hours=LOOKBACK_HOURS)
    since_iso = since.strftime("%Y-%m-%dT%H:%M:%SZ")
    leads = soql_all("SELECT Id, Portal_ID__c, IsConverted FROM Lead WHERE LastModifiedDate >= " + since_iso + " AND Portal_ID__c != null")
    wp = [r for r in leads if r.get("Portal_ID__c") not in (None, "")]
    ids = [r["Id"] for r in wp]
    portals = [r["Portal_ID__c"] for r in wp]
    byLead = set(); byPortal = set()
    for ch in batched(ids):
        lits = "','".join(soql_lit(x) for x in ch)
        for r in soql_all("SELECT Lead__c FROM Org__c WHERE Lead__c IN ('" + lits + "')"):
            byLead.add(r.get("Lead__c"))
    for ch in batched(portals):
        lits = "','".join(soql_lit(x) for x in ch)
        for r in soql_all("SELECT ID__c FROM Org__c WHERE ID__c IN ('" + lits + "')"):
            byPortal.add(r.get("ID__c"))
    return [r for r in wp if r["Id"] not in byLead and r["Portal_ID__c"] not in byPortal], len(wp)

def orgs_present(pairs):
    """Given (lead_id, portal) pairs, return (have_by_lead, have_by_portal) sets
    of those that now have an Org__c (matched on either key), batched."""
    have_lead = set(); have_portal = set()
    for ch in batched([p[0] for p in pairs]):
        lits = "','".join(soql_lit(x) for x in ch)
        for r in soql_all("SELECT Lead__c FROM Org__c WHERE Lead__c IN ('" + lits + "')"):
            have_lead.add(r.get("Lead__c"))
    for ch in batched([p[1] for p in pairs]):
        lits = "','".join(soql_lit(x) for x in ch)
        for r in soql_all("SELECT ID__c FROM Org__c WHERE ID__c IN ('" + lits + "')"):
            have_portal.add(r.get("ID__c"))
    return have_lead, have_portal

def verify_orgs(pairs, rounds=4, gap=15):
    """Return the subset of attempted (lead_id, portal) pairs that still have no
    Org__c after up to `rounds` re-checks spaced `gap`s apart. Bounded retries
    absorb the flow's async settle; whatever remains is a genuine unhealed gap."""
    pending = list(pairs)
    for _ in range(rounds):
        if not pending:
            break
        time.sleep(gap)
        hl, hp = orgs_present(pending)
        pending = [(lid, portal) for (lid, portal) in pending if lid not in hl and portal not in hp]
    return pending

def duplicate_portals(pairs):
    """Portals among `pairs` that now have more than one Org__c. Org__c.ID__c is
    not unique, so a duplicate is possible if the async flow ever gets fired
    twice; surface it rather than let it pass silently. Reported, never deleted -
    removing records is a destructive operation and stays a human decision."""
    seen = {}
    for ch in batched([p[1] for p in pairs]):
        lits = "','".join(soql_lit(x) for x in ch)
        for r in soql_all("SELECT ID__c FROM Org__c WHERE ID__c IN ('" + lits + "')"):
            k = r.get("ID__c")
            seen[k] = seen.get(k, 0) + 1
    return [k for k, n in seen.items() if n > 1]

def split_fixable(missing):
    """Converted Leads are read-only: the Portal_ID toggle can never commit, so
    re-trying them every run is pure spin. Keep them counted (and, only for local
    ops, listed) instead of silently filtering, so a real gap is still reported."""
    fixable = [r for r in missing if not r.get("IsConverted")]
    converted = [r for r in missing if r.get("IsConverted")]
    return fixable, converted

def backfill_one(lid, portal):
    """Re-trigger the Org-create flow for one Lead. Returns one of:
    'done', 'skip', 'error', 'anomalous'.

    This writes only Reconcile_Touch__c, but that is still a Lead SAVE, so it
    re-runs the same Lead automation any other Lead save would (see the module
    docstring for the audited list). No existing business value can be
    overwritten; the only effect is fill-if-null defaulting happening sooner."""
    if not PORTAL_RE.match(str(portal)):
        return "anomalous"
    lit_lid = soql_lit(lid)
    lit_portal = soql_lit(portal)
    # Dedup guard: an Org may have appeared since the scan.
    if soql_all("SELECT Id FROM Org__c WHERE Lead__c='" + lit_lid + "' OR ID__c='" + lit_portal + "'"):
        return "skip"
    # Capture the current trigger-field value so an ambiguous write can be
    # resolved by re-reading instead of blindly resubmitting the save.
    cur = soql_all("SELECT " + RECONCILE_FIELD + " FROM Lead WHERE Id='" + lit_lid + "'")
    if not cur:
        return "skip"
    res, detail = trigger_touch(lid, cur[0].get(RECONCILE_FIELD))
    if res == "error":
        log("PATCH_ERROR" + ((" lead=" + lid) if LOG_IDS else "")
            + ((" " + detail[:200]) if LOG_IDS else ""))
    return res

def main():
    # Fail closed: the trigger field ships via an L2 pack change that must be
    # deployed to the target org first. This workflow is intended to be enabled
    # only after that deploy, so a missing field means a broken/rolled-back pack
    # deployment - a real problem the run must surface (red), never a silent
    # green while no Leads get repaired.
    if not field_exists(RECONCILE_FIELD):
        log("ERROR trigger field " + RECONCILE_FIELD + " missing in target org (pack not deployed / rolled back)")
        return 1
    missing, total_portal = find_missing()
    fixable, converted = split_fixable(missing)
    log("SCAN key=LastModified lookback=" + str(LOOKBACK_HOURS) + "h portal_leads=" + str(total_portal)
        + " missing_org=" + str(len(missing)) + " fixable=" + str(len(fixable))
        + " converted_readonly=" + str(len(converted)))
    # Converted Leads with a Portal but no Org are a real, unrepaired gap: they
    # are locked (Lead lockRule IsConverted=true), so re-triggering the flow can
    # never create their Org - only a manual Org create (as for the BIXPAYMENTS
    # case) resolves them. This job does not replicate the flow's Org mapping, so
    # it fails closed on them: a run is NEVER green while such a gap persists.
    converted_gap = len(converted)
    if converted:
        detail = ""
        if LOG_IDS:
            detail = ": " + ", ".join(r["Id"] + "/" + str(r["Portal_ID__c"]) for r in converted[:20])
        log("UNFIXABLE converted_readonly=" + str(converted_gap) + " (locked; need manual Org create)" + detail)
    if not fixable:
        log("no fixable candidates this run")
        return 1 if converted_gap else 0
    # Collapse candidates to one Lead per Portal for this run. Two Leads sharing
    # the same missing Portal would otherwise both fire the async Org flow and,
    # because Org__c.ID__c is not unique, race into a duplicate Org. Touching one
    # per Portal closes that in-run vector; the deferred siblings are picked up on
    # a later run once the Org exists (the dedup guard then skips them).
    seen_portal = set()
    deduped = []
    same_portal = 0
    for c in fixable:
        p = c["Portal_ID__c"]
        if p in seen_portal:
            same_portal += 1
            continue
        seen_portal.add(p)
        deduped.append(c)
    if same_portal:
        log("PORTAL_DEDUP deferred=" + str(same_portal) + " (multiple Leads share one missing Portal this run)")
    todo = deduped[:MAX_PER_RUN]
    if len(deduped) > MAX_PER_RUN:
        log("CAP backfilling " + str(MAX_PER_RUN) + " of " + str(len(deduped)) + " this run")
    counts = {"done": 0, "skip": 0, "error": 0, "anomalous": 0}
    attempted = []
    for c in todo:
        res = backfill_one(c["Id"], c["Portal_ID__c"])
        counts[res] += 1
        if res == "done":
            attempted.append((c["Id"], c["Portal_ID__c"]))
    # Confirm the flow actually created an Org for each Lead we touched. A
    # successful PATCH only proves the trigger fired, not that Get_Org produced
    # the record, so re-verify the attempted set (with bounded retries to absorb
    # async settle) and treat anything still missing as a real, unhealed gap.
    unresolved = verify_orgs(attempted)
    dups = duplicate_portals(attempted) if attempted else []
    log("BACKFILL touched=" + str(counts["done"]) + " skipped=" + str(counts["skip"])
        + " errors=" + str(counts["error"]) + " anomalous=" + str(counts["anomalous"])
        + " unresolved_attempted=" + str(len(unresolved)) + " duplicate_portals=" + str(len(dups)))
    if unresolved and LOG_IDS:
        log("UNRESOLVED " + ", ".join(lid + "/" + str(portal) for lid, portal in unresolved[:20]))
    if dups:
        log("DUPLICATE_ORG portals=" + str(len(dups)) + " (needs manual review)"
            + ((": " + ", ".join(dups[:20])) if LOG_IDS else ""))
    # Fail the run (turn CI red) on any real, non-transient gap: candidates we
    # attempted but could not confirm an Org for, PATCH errors, anomalous
    # (non-numeric) Portal IDs, converted/locked Leads that need a manual Org, or
    # duplicate Orgs we may have contributed to. Only a fully clean run stays
    # green, so a persisting safety-net gap can never hide behind it.
    return 1 if (counts["error"] or counts["anomalous"] or unresolved
                 or converted_gap or dups) else 0

if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as e:
        log("ERROR " + type(e).__name__ + ": " + str(e)); sys.exit(1)