#!/usr/bin/env python3
"""
Synchronize all user and role grants from the Delphi Snowflake account to Orchard.

For every role and every user in Delphi, runs SHOW GRANTS TO ROLE / TO USER and
SHOW FUTURE GRANTS TO ROLE, reconstructs each grant as a GRANT statement, then
either applies them to Orchard or writes them to a SQL file.

Grants whose granted-on database does not exist in Orchard are skipped. Per-grant
failures during execution are aggregated and printed at the end so a single bad
statement doesn't abort the run.

Usage:
  pip install snowflake-connector-python
  python3 sync-snowflake-grants.py                          # apply to Orchard
  python3 sync-snowflake-grants.py --output-file grants.sql # write SQL only
  python3 sync-snowflake-grants.py --entity MY_ROLE         # restrict to one entity

Authentication:
  Uses key-pair auth if SNOWFLAKE_PRIVATE_KEY_PATH is set, otherwise falls back
  to externalbrowser SSO. Per-account overrides via ORCHARD_* / DELPHI_* prefixes.

Required env vars:
  SNOWFLAKE_USER   - your Snowflake username

Optional env vars:
  SNOWFLAKE_PRIVATE_KEY_PATH         - path to PEM private key file
  SNOWFLAKE_PRIVATE_KEY_PASSPHRASE   - passphrase if key is encrypted
  SNOWFLAKE_ROLE                     - role to use (default: SECURITYADMIN)
  ORCHARD_SNOWFLAKE_USER / DELPHI_SNOWFLAKE_USER   - per-account user override
  ORCHARD_SNOWFLAKE_ROLE / DELPHI_SNOWFLAKE_ROLE   - per-account role override
  ORCHARD_ACCOUNT / DELPHI_ACCOUNT                 - account identifier overrides
  ORCHARD_SNOWFLAKE_WAREHOUSE / DELPHI_SNOWFLAKE_WAREHOUSE - warehouse overrides
"""

import argparse
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from typing import Any

try:
    import snowflake.connector
except ImportError:
    sys.exit(
        "snowflake-connector-python is required.\n"
        "Install it with: pip install snowflake-connector-python"
    )


# Databases whose grants should always be excluded, even if they exist in Orchard.
# Names are matched case-insensitively. Append to this list as more exclusions
# are identified.
EXCLUDED_DATABASES: set[str] = {
    "CHARTMETRIC", # Imported DB - won't be replicated
    "DEV_ENGINEERING", # DEV_ENGINEERING will be renamed in Delphi
    "ORCHARD_APP_REPORTING", # This will be excluded from replication
    "ORCHARD_APP_REPORTING_V2", # This will be renamed in Delphi
    "SIGMA", # This will be renamed in Delphi
    "SME_ANALYTICS", # This will be renamed in Orchard
    "SNOWFLAKE", # This will be excluded from replication, grants are made via GRANT IMPORTED PRIVILEGES
    "SNOWFLAKE_LEARNING_DB", # Also excluded from replcation
}

# Database name prefixes whose grants should always be excluded. Matched
# case-insensitively against the database segment. USER$ covers Snowflake's
# internal per-user databases.
EXCLUDED_DATABASE_PREFIXES: tuple[str, ...] = (
    "USER$",
)

# Grantee roles to exclude — any grant whose recipient is one of these roles is
# skipped, and we don't run SHOW GRANTS against them. ORGADMIN is the
# org-level system role that shouldn't be touched by an account-level sync.
EXCLUDED_GRANTEE_ROLES: set[str] = {
    "AUTO_FULFILLMENT_EXECUTOR", # internal system role
    "ORGADMIN",
}

# Prefixes for roles that should never appear as the role being granted in a
# GRANT ROLE statement. USER$ covers Snowflake's internal per-user roles,
# which are auto-managed and can't be re-granted.
EXCLUDED_GRANTED_ROLE_PREFIXES: tuple[str, ...] = (
    "USER$",
)

# granted_on object types whose grants are always skipped regardless of name.
# COMPUTE_POOL: account-specific resources, not replicated.
EXCLUDED_GRANTED_ON_TYPES: set[str] = {
    "COMPUTE_POOL",
    "CONNECTION",
    "DATA_EXCHANGE_LISTING",
    "REPLICATION_GROUP",
    "SHARE",
}

# Integration name prefixes whose grants are skipped. Matched case-insensitively
# against the integration name. SNOWSERVICE- covers Snowflake's auto-generated
# Snowpark Container Services integrations.
EXCLUDED_INTEGRATION_PREFIXES: tuple[str, ...] = (
    "SNOWSERVICE-",
)


# ------------------------------------------------------------------ connection

def _private_key_bytes() -> bytes | None:
    path = os.environ.get("SNOWFLAKE_PRIVATE_KEY_PATH")
    if not path:
        return None
    from cryptography.hazmat.backends import default_backend
    from cryptography.hazmat.primitives import serialization

    passphrase_str = os.environ.get("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE")
    passphrase = passphrase_str.encode() if passphrase_str else None
    with open(path, "rb") as fh:
        private_key = serialization.load_pem_private_key(
            fh.read(), password=passphrase, backend=default_backend()
        )
    return private_key.private_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )


def _conn_params(account_name: str, env_prefix: str, default_warehouse: str) -> dict[str, Any]:
    def env(key: str) -> str | None:
        return os.environ.get(f"{env_prefix}{key}") or os.environ.get(key)

    params: dict[str, Any] = {
        "account": account_name,
        "user": env("SNOWFLAKE_USER"),
        "role": env("SNOWFLAKE_ROLE") or "SECURITYADMIN",
        "warehouse": env("SNOWFLAKE_WAREHOUSE") or default_warehouse,
    }

    pk_bytes = _private_key_bytes()
    if pk_bytes:
        params["private_key"] = pk_bytes
        params["authenticator"] = "snowflake_jwt"
    else:
        params["authenticator"] = "externalbrowser"

    return params


# ------------------------------------------------------------------ identifier quoting

def quote_ident(name: str) -> str:
    """Wrap an identifier in double quotes, escaping embedded quotes."""
    return '"' + name.replace('"', '""') + '"'


def quote_qualified(name: str) -> str:
    """Quote each dot-separated segment independently and rejoin."""
    return ".".join(quote_ident(seg) for seg in name.split("."))


def quote_callable(name: str) -> str:
    """Quote a callable's qualified name, leaving its signature outside the quotes.
    Input  "DB.SCHEMA.FN(TABLE(NUMBER))"  →  "DB"."SCHEMA"."FN"(TABLE(NUMBER))
    """
    paren = name.find("(")
    if paren < 0:
        return quote_qualified(name)
    return quote_qualified(name[:paren]) + name[paren:]


# granted_on values whose `name` includes a parenthesized signature.
CALLABLE_OBJECT_TYPES = {"FUNCTION", "PROCEDURE", "EXTERNAL FUNCTION"}


# ------------------------------------------------------------------ row fetching

def _row_lower(row: dict[str, Any]) -> dict[str, Any]:
    """Normalize dict keys to lowercase — SHOW commands return lowercase but
    we don't want to depend on driver behavior."""
    return {k.lower(): v for k, v in row.items()}


def _list_show(cur: Any, sql: str) -> list[dict[str, Any]]:
    cur.execute(sql)
    return [_row_lower(r) for r in cur.fetchall()]


def fetch_account_entities(
    conn_params: dict[str, Any],
) -> tuple[list[str], list[str]]:
    """Return (role_names, user_names) via SHOW ROLES / SHOW USERS.
    Filters SHOW ROLES output to kind == 'ROLE' (excludes APPLICATION/DATABASE roles)."""
    conn = snowflake.connector.connect(**conn_params)
    try:
        cur = conn.cursor(snowflake.connector.DictCursor)
        roles_rows = _list_show(cur, "SHOW ROLES")
        users_rows = _list_show(cur, "SHOW USERS")
    finally:
        conn.close()

    roles: list[str] = []
    for r in roles_rows:
        kind = (r.get("kind") or "ROLE").upper()
        if kind == "ROLE":
            roles.append(r["name"])

    users: list[str] = [r["name"] for r in users_rows]
    return sorted(roles), sorted(users)


def fetch_orchard_databases(conn_params: dict[str, Any]) -> set[str]:
    conn = snowflake.connector.connect(**conn_params)
    try:
        cur = conn.cursor(snowflake.connector.DictCursor)
        rows = _list_show(cur, "SHOW DATABASES")
    finally:
        conn.close()
    return {r["name"].upper() for r in rows}


def _fetch_grants_for_role(
    conn_params: dict[str, Any], role: str
) -> tuple[str, list[dict[str, Any]], list[dict[str, Any]]]:
    """Run SHOW GRANTS TO ROLE and SHOW FUTURE GRANTS TO ROLE on a fresh connection."""
    conn = snowflake.connector.connect(**conn_params)
    try:
        cur = conn.cursor(snowflake.connector.DictCursor)
        ident = quote_ident(role)
        grants = _list_show(cur, f"SHOW GRANTS TO ROLE {ident}")
        future = _list_show(cur, f"SHOW FUTURE GRANTS TO ROLE {ident}")
    finally:
        conn.close()
    return role, grants, future


def _fetch_grants_for_user(
    conn_params: dict[str, Any], user: str
) -> tuple[str, list[dict[str, Any]]]:
    conn = snowflake.connector.connect(**conn_params)
    try:
        cur = conn.cursor(snowflake.connector.DictCursor)
        ident = quote_ident(user)
        grants = _list_show(cur, f"SHOW GRANTS TO USER {ident}")
    finally:
        conn.close()
    return user, grants


def fetch_all_grants(
    conn_params: dict[str, Any],
    roles: list[str],
    users: list[str],
    max_workers: int,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]:
    """Return (role_grants, future_role_grants, user_grants) — flat lists of rows."""
    role_grants: list[dict[str, Any]] = []
    future_role_grants: list[dict[str, Any]] = []
    user_grants: list[dict[str, Any]] = []

    total = len(roles) + len(users)
    print(
        f"Fetching grants in parallel "
        f"({len(roles):,} roles + {len(users):,} users, {max_workers} workers)…",
        flush=True,
    )

    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = []
        for r in roles:
            futures.append(pool.submit(_fetch_grants_for_role, conn_params, r))
        for u in users:
            futures.append(pool.submit(_fetch_grants_for_user, conn_params, u))

        done = 0
        for fut in as_completed(futures):
            result = fut.result()
            if len(result) == 3:
                _name, grants, future_grants = result
                role_grants.extend(grants)
                future_role_grants.extend(future_grants)
            else:
                _name, grants = result
                user_grants.extend(grants)
            done += 1
            if done % 50 == 0 or done == total:
                print(f"  …{done}/{total}", flush=True)

    print(
        f"  → {len(role_grants):,} role grants, "
        f"{len(future_role_grants):,} future role grants, "
        f"{len(user_grants):,} user grants",
        flush=True,
    )
    return role_grants, future_role_grants, user_grants


# ------------------------------------------------------------------ object-type pluralization

# Snowflake future-grant object types; pluralized for the IN ... clause.
_PLURAL_OBJECT_TYPES = {
    "TABLE": "TABLES",
    "EXTERNAL TABLE": "EXTERNAL TABLES",
    "DYNAMIC TABLE": "DYNAMIC TABLES",
    "ICEBERG TABLE": "ICEBERG TABLES",
    "VIEW": "VIEWS",
    "MATERIALIZED VIEW": "MATERIALIZED VIEWS",
    "SCHEMA": "SCHEMAS",
    "STAGE": "STAGES",
    "FILE FORMAT": "FILE FORMATS",
    "SEQUENCE": "SEQUENCES",
    "PIPE": "PIPES",
    "STREAM": "STREAMS",
    "TASK": "TASKS",
    "FUNCTION": "FUNCTIONS",
    "PROCEDURE": "PROCEDURES",
    "MASKING POLICY": "MASKING POLICIES",
    "ROW ACCESS POLICY": "ROW ACCESS POLICIES",
    "PASSWORD POLICY": "PASSWORD POLICIES",
    "SESSION POLICY": "SESSION POLICIES",
    "TAG": "TAGS",
    "EVENT TABLE": "EVENT TABLES",
    "MODEL": "MODELS",
    "DATASET": "DATASETS",
    "CORTEX SEARCH SERVICE": "CORTEX SEARCH SERVICES",
}


def _pluralize_object_type(obj_type: str) -> str:
    obj_type = obj_type.upper().strip()
    if obj_type in _PLURAL_OBJECT_TYPES:
        return _PLURAL_OBJECT_TYPES[obj_type]
    if obj_type.endswith("Y"):
        return obj_type[:-1] + "IES"
    if obj_type.endswith("S"):
        return obj_type
    return obj_type + "S"


# ------------------------------------------------------------------ grant keys

# Normalized keys identify a grant independently of which account it came from.
# Used to filter Delphi grants that already exist in Orchard.
GrantKey = tuple

def role_grant_key(row: dict[str, Any]) -> GrantKey:
    granted_on = (row.get("granted_on") or "").upper()
    name = (row.get("name") or "").upper()
    # ACCOUNT-level grants put the account identifier in `name`, which differs
    # between Delphi and Orchard. The rendered SQL ignores `name` for ACCOUNT
    # (`GRANT ... ON ACCOUNT TO ROLE X`), so we drop it from the key too.
    if granted_on == "ACCOUNT":
        name = ""
    return (
        "role",
        (row.get("grantee_name") or "").upper(),
        (row.get("privilege") or "").upper(),
        granted_on,
        name,
        str(row.get("grant_option") or "").lower() == "true",
    )


def future_role_grant_key(row: dict[str, Any]) -> GrantKey:
    return (
        "future_role",
        (row.get("grantee_name") or "").upper(),
        (row.get("privilege") or "").upper(),
        (row.get("grant_on") or row.get("granted_on") or "").upper(),
        (row.get("name") or "").upper(),
        str(row.get("grant_option") or "").lower() == "true",
    )


def user_grant_key(row: dict[str, Any]) -> GrantKey:
    return (
        "user",
        (row.get("grantee_name") or "").upper(),
        (row.get("role") or "").upper(),
    )


# Sort keys group statements by (grant type, target object, privilege, grantee)
# for deterministic, readable output. Privilege is a sub-key within each target
# so e.g. all MONITOR grants on a warehouse are listed contiguously, then all
# OPERATE, then OWNERSHIP, etc. — rather than interleaved by grantee.
# Distinct from GrantKey, which is for cross-account dedup.
SortKey = tuple[str, str, str, str]


def role_grant_sort_key(row: dict[str, Any]) -> SortKey:
    return (
        (row.get("granted_on") or "").upper(),
        (row.get("name") or "").upper(),
        (row.get("privilege") or "").upper(),
        (row.get("grantee_name") or "").upper(),
    )


def future_role_grant_sort_key(row: dict[str, Any]) -> SortKey:
    grant_on = (row.get("grant_on") or row.get("granted_on") or "").upper()
    return (
        f"FUTURE {grant_on}",
        (row.get("name") or "").upper(),
        (row.get("privilege") or "").upper(),
        (row.get("grantee_name") or "").upper(),
    )


def user_grant_sort_key(row: dict[str, Any]) -> SortKey:
    return (
        "ROLE TO USER",
        (row.get("role") or "").upper(),
        "",
        (row.get("grantee_name") or "").upper(),
    )


# ------------------------------------------------------------------ row → SQL

# (sql, db_required) — db_required is None if no database scope applies.
RenderResult = tuple[str | None, str | None]


def render_user_grant(row: dict[str, Any]) -> RenderResult:
    """SHOW GRANTS TO USER row → GRANT ROLE x TO USER y;"""
    role = row.get("role")
    grantee = row.get("grantee_name") or row.get("grantee")
    if not role or not grantee:
        return None, None
    sql = f"GRANT ROLE {quote_ident(role)} TO USER {quote_ident(grantee)};"
    return sql, None


def render_role_grant(row: dict[str, Any]) -> RenderResult:
    """SHOW GRANTS TO ROLE row → appropriate GRANT statement, with db_required."""
    privilege = (row.get("privilege") or "").upper()
    granted_on = (row.get("granted_on") or "").upper()
    name = row.get("name") or ""
    grantee_name = row.get("grantee_name") or ""
    grant_option = str(row.get("grant_option") or "").lower() == "true"

    if not privilege or not granted_on or not grantee_name:
        return None, None

    grantee_sql = quote_ident(grantee_name)
    suffix_parts: list[str] = []
    if privilege == "OWNERSHIP":
        suffix_parts.append("COPY CURRENT GRANTS")
    elif grant_option:
        suffix_parts.append("WITH GRANT OPTION")
    suffix = (" " + " ".join(suffix_parts)) if suffix_parts else ""

    if granted_on == "ROLE":
        # GRANT ROLE <name> TO ROLE <grantee>; — privilege is USAGE for role-to-role.
        # Some "ROLE" rows from Snowflake have qualified names (e.g. Streamlit
        # application roles like "DB.SCHEMA.APP.ROLE"); treat those as
        # database-scoped so the EXCLUDED_DATABASES filter applies.
        is_qualified = "." in name
        quoted_name = quote_qualified(name) if is_qualified else quote_ident(name)
        db = name.split(".", 1)[0] if is_qualified else None
        if privilege == "OWNERSHIP":
            sql = f"GRANT OWNERSHIP ON ROLE {quoted_name} TO ROLE {grantee_sql}{suffix};"
        else:
            sql = f"GRANT ROLE {quoted_name} TO ROLE {grantee_sql};"
        return sql, db

    if granted_on == "ACCOUNT":
        sql = f"GRANT {privilege} ON ACCOUNT TO ROLE {grantee_sql}{suffix};"
        return sql, None

    if granted_on == "DATABASE":
        sql = f"GRANT {privilege} ON DATABASE {quote_ident(name)} TO ROLE {grantee_sql}{suffix};"
        return sql, name

    # DATABASE ROLE names look like "<DB>.<ROLE>" so they're database-scoped.
    if granted_on == "DATABASE ROLE":
        db = name.split(".", 1)[0]
        sql = (
            f"GRANT {privilege} ON DATABASE ROLE {quote_qualified(name)} "
            f"TO ROLE {grantee_sql}{suffix};"
        )
        return sql, db

    # Account-scoped objects (no database parent).
    account_scoped = {
        "WAREHOUSE", "INTEGRATION", "USER", "RESOURCE MONITOR", "NETWORK POLICY",
        "CONNECTION", "SHARE", "FAILOVER GROUP", "REPLICATION GROUP",
        "EXTERNAL VOLUME", "COMPUTE POOL", "WAREHOUSE POOL", "API INTEGRATION",
        "SECURITY INTEGRATION", "STORAGE INTEGRATION", "CATALOG INTEGRATION",
        "NOTIFICATION INTEGRATION", "ORGANIZATION", "APPLICATION",
        "APPLICATION PACKAGE",
    }
    if granted_on in account_scoped:
        sql = (
            f"GRANT {privilege} ON {granted_on} {quote_ident(name)} "
            f"TO ROLE {grantee_sql}{suffix};"
        )
        return sql, None

    # SCHEMA, TABLE, VIEW, STAGE, FUNCTION, etc. — name is qualified, first segment is db.
    db = name.split(".", 1)[0] if "." in name else None
    if granted_on in CALLABLE_OBJECT_TYPES:
        # FUNCTION / PROCEDURE names include a signature like "(TABLE(NUMBER))"
        # that must sit outside the quoted identifier.
        quoted_name = quote_callable(name)
    else:
        quoted_name = quote_qualified(name)
    sql = (
        f"GRANT {privilege} ON {granted_on} {quoted_name} "
        f"TO ROLE {grantee_sql}{suffix};"
    )
    return sql, db


def render_future_role_grant(row: dict[str, Any]) -> RenderResult:
    """SHOW FUTURE GRANTS TO ROLE row → GRANT ... ON FUTURE <plural> IN ... statement."""
    privilege = (row.get("privilege") or "").upper()
    grant_on = (row.get("grant_on") or row.get("granted_on") or "").upper()
    name = row.get("name") or ""
    grantee_name = row.get("grantee_name") or ""
    grant_option = str(row.get("grant_option") or "").lower() == "true"

    if not privilege or not grant_on or not name or not grantee_name:
        return None, None

    # name is e.g. "DB.SCHEMA.<TABLE>" or "DB.<SCHEMA>". Last segment marks the
    # object type (often wrapped in <>). Strip and pluralize.
    segments = name.split(".")
    if len(segments) < 2:
        return None, None
    last = segments[-1].strip("<>")
    plural = _pluralize_object_type(last or grant_on)
    scope_segments = segments[:-1]

    if len(scope_segments) == 1:
        scope_clause = f"IN DATABASE {quote_ident(scope_segments[0])}"
        db = scope_segments[0]
    elif len(scope_segments) == 2:
        scope_clause = f"IN SCHEMA {quote_qualified('.'.join(scope_segments))}"
        db = scope_segments[0]
    else:
        # Unexpected shape — bail rather than emit malformed SQL.
        return None, None

    suffix = " WITH GRANT OPTION" if grant_option else ""
    grantee_sql = quote_ident(grantee_name)
    sql = (
        f"GRANT {privilege} ON FUTURE {plural} {scope_clause} "
        f"TO ROLE {grantee_sql}{suffix};"
    )
    return sql, db


# ------------------------------------------------------------------ apply / write

def write_to_file(
    path: str,
    statements: list[str],
    delphi_account: str,
    orchard_account: str,
) -> None:
    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    with open(path, "w") as fh:
        fh.write(f"-- Grant sync: {delphi_account} → {orchard_account}\n")
        fh.write(f"-- Generated: {timestamp}\n")
        fh.write(f"-- Statements: {len(statements)}\n\n")
        for stmt in statements:
            fh.write(stmt + "\n")


def write_failures_to_file(
    path: str,
    failures: list[tuple[str, str]],
    delphi_account: str,
    orchard_account: str,
) -> None:
    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    with open(path, "w") as fh:
        fh.write(f"-- Failed grants from sync: {delphi_account} → {orchard_account}\n")
        fh.write(f"-- Generated: {timestamp}\n")
        fh.write(f"-- Failures: {len(failures)}\n\n")
        for sql, err in failures:
            fh.write(f"-- ERROR: {err}\n")
            fh.write(sql + "\n\n")


def apply_to_orchard(
    conn_params: dict[str, Any],
    statements: list[str],
) -> tuple[int, list[tuple[str, str]]]:
    applied = 0
    failures: list[tuple[str, str]] = []
    if not statements:
        return applied, failures

    print(f"Applying {len(statements):,} statements to Orchard…", flush=True)
    conn = snowflake.connector.connect(**conn_params)
    try:
        cur = conn.cursor()
        for i, stmt in enumerate(statements, 1):
            try:
                cur.execute(stmt)
                applied += 1
            except Exception as exc:
                failures.append((stmt, str(exc).strip()))
            if i % 100 == 0 or i == len(statements):
                print(f"  …{i}/{len(statements)} ({len(failures)} failed)", flush=True)
    finally:
        conn.close()
    return applied, failures


# ------------------------------------------------------------------ main

def main() -> None:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--output-file",
        metavar="PATH",
        help="Write GRANT statements to this file instead of executing against Orchard.",
    )
    parser.add_argument(
        "--failures-file",
        metavar="PATH",
        default="failed-grants.sql",
        help="In execute mode, write any failed GRANT statements to this file "
             "(default: failed-grants.sql). File is only written if there are failures.",
    )
    parser.add_argument(
        "--entity",
        metavar="NAME",
        help="Restrict to a single role or user name (case-sensitive).",
    )
    parser.add_argument(
        "--max-workers",
        type=int,
        default=20,
        help="Parallel workers for SHOW GRANTS fetching (default: 20).",
    )
    ownership_group = parser.add_mutually_exclusive_group()
    ownership_group.add_argument(
        "--no-ownership",
        dest="include_ownership",
        action="store_false",
        help="Exclude OWNERSHIP grants from the output.",
    )
    ownership_group.add_argument(
        "--include-ownership",
        dest="include_ownership",
        action="store_true",
        help="Include OWNERSHIP grants (default).",
    )
    parser.set_defaults(include_ownership=True)
    parser.add_argument(
        "--orchard-account",
        default=os.environ.get("ORCHARD_ACCOUNT", "sme-orchard"),
        help="Orchard account identifier (default: sme-orchard)",
    )
    parser.add_argument(
        "--delphi-account",
        default=os.environ.get("DELPHI_ACCOUNT", "sme-delphi"),
        help="Delphi account identifier (default: sme-delphi)",
    )
    args = parser.parse_args()

    delphi_params = _conn_params(args.delphi_account, "DELPHI_", "DEV_OWS_WH")
    orchard_params = _conn_params(args.orchard_account, "ORCHARD_", "DEV_OWS_WAREHOUSE")

    # Phase 1 — enumerate Delphi roles and users
    print(f"Connecting to Delphi ({args.delphi_account})…", flush=True)
    roles, users = fetch_account_entities(delphi_params)
    print(f"  → {len(roles):,} roles, {len(users):,} users", flush=True)

    if args.entity:
        target = args.entity
        roles = [r for r in roles if r == target]
        users = [u for u in users if u == target]
        if not roles and not users:
            sys.exit(f"Entity '{target}' not found among Delphi roles or users.")

    excluded_grantees_upper = {r.upper() for r in EXCLUDED_GRANTEE_ROLES}
    roles = [r for r in roles if r.upper() not in excluded_grantees_upper]

    # Phase 2 — fetch all Delphi grants in parallel
    role_grants, future_role_grants, user_grants = fetch_all_grants(
        delphi_params, roles, users, args.max_workers
    )

    # Phase 3 — enumerate Orchard entities + databases, and fetch existing
    # Orchard grants for entities that exist in both accounts. Entities only in
    # Delphi have no Orchard grants to diff against — their grants will be
    # emitted as-is.
    print(f"Connecting to Orchard ({args.orchard_account})…", flush=True)
    orchard_roles, orchard_users = fetch_account_entities(orchard_params)
    orchard_dbs = fetch_orchard_databases(orchard_params)
    print(
        f"  → {len(orchard_roles):,} roles, {len(orchard_users):,} users, "
        f"{len(orchard_dbs):,} databases",
        flush=True,
    )

    orchard_role_set = set(orchard_roles)
    orchard_user_set = set(orchard_users)
    shared_roles = [r for r in roles if r in orchard_role_set]
    shared_users = [u for u in users if u in orchard_user_set]

    o_role_grants, o_future_role_grants, o_user_grants = fetch_all_grants(
        orchard_params, shared_roles, shared_users, args.max_workers
    )

    orchard_keys: set[GrantKey] = set()
    for row in o_role_grants:
        orchard_keys.add(role_grant_key(row))
    for row in o_future_role_grants:
        orchard_keys.add(future_role_grant_key(row))
    for row in o_user_grants:
        orchard_keys.add(user_grant_key(row))

    # Phase 4+5 — render and filter
    pending: list[tuple[SortKey, str]] = []
    seen_statements: set[str] = set()
    skipped_missing_db: list[tuple[str, str]] = []  # (db, sql)
    skipped_excluded_db: list[tuple[str, str]] = []  # (db, sql)
    skipped_already_in_orchard = 0
    skipped_duplicate = 0
    skipped_excluded_grantee = 0
    skipped_excluded_granted_role = 0
    skipped_excluded_object_type = 0
    skipped_excluded_integration = 0
    skipped_ownership = 0
    unrenderable = 0
    excluded_upper = {db.upper() for db in EXCLUDED_DATABASES}
    excluded_prefixes_upper = tuple(p.upper() for p in EXCLUDED_DATABASE_PREFIXES)
    excluded_granted_role_prefixes_upper = tuple(
        p.upper() for p in EXCLUDED_GRANTED_ROLE_PREFIXES
    )
    excluded_object_types_upper = {t.upper() for t in EXCLUDED_GRANTED_ON_TYPES}
    excluded_integration_prefixes_upper = tuple(
        p.upper() for p in EXCLUDED_INTEGRATION_PREFIXES
    )

    def consider(result: RenderResult, key: GrantKey, sort_key: SortKey) -> None:
        nonlocal unrenderable, skipped_already_in_orchard, skipped_duplicate
        nonlocal skipped_excluded_grantee, skipped_excluded_granted_role
        nonlocal skipped_ownership, skipped_excluded_object_type
        nonlocal skipped_excluded_integration
        sql, db_required = result
        if not sql:
            unrenderable += 1
            return
        # Excluded granted_on object type (e.g. COMPUTE_POOL).
        # role/future_role keys carry granted_on at index 3.
        if (
            key[0] in ("role", "future_role")
            and len(key) > 3
            and isinstance(key[3], str)
            and key[3] in excluded_object_types_upper
        ):
            skipped_excluded_object_type += 1
            return
        # Excluded INTEGRATION names by prefix (e.g. SNOWSERVICE-…). name is
        # at index 4 for role/future_role keys. Snowflake sometimes stores
        # integration names with literal embedded double-quotes, so strip any
        # leading quotes before the prefix check.
        if (
            key[0] in ("role", "future_role")
            and len(key) > 4
            and isinstance(key[3], str)
            and key[3] == "INTEGRATION"
            and isinstance(key[4], str)
            and key[4].lstrip('"').startswith(excluded_integration_prefixes_upper)
        ):
            skipped_excluded_integration += 1
            return
        # OWNERSHIP filter (only role/future_role rows carry a privilege at key[2]).
        if (
            not args.include_ownership
            and key[0] in ("role", "future_role")
            and len(key) > 2
            and key[2] == "OWNERSHIP"
        ):
            skipped_ownership += 1
            return
        # key shape: (kind, grantee, ...) — index 1 is always the grantee.
        if len(key) > 1 and isinstance(key[1], str) and key[1] in excluded_grantees_upper:
            skipped_excluded_grantee += 1
            return
        # Skip GRANT ROLE statements where the role being granted matches an
        # excluded prefix. Two cases: SHOW GRANTS TO USER (key=("user",
        # grantee, role)) and SHOW GRANTS TO ROLE rows with granted_on=ROLE
        # (key=("role", grantee, privilege, "ROLE", name, grant_option)).
        granted_role: str | None = None
        if key[0] == "user" and len(key) >= 3 and isinstance(key[2], str):
            granted_role = key[2]
        elif (
            key[0] == "role"
            and len(key) >= 5
            and key[3] == "ROLE"
            and isinstance(key[4], str)
        ):
            granted_role = key[4]
        if granted_role and granted_role.startswith(excluded_granted_role_prefixes_upper):
            skipped_excluded_granted_role += 1
            return
        if key in orchard_keys:
            skipped_already_in_orchard += 1
            return
        if db_required:
            db_upper = db_required.upper()
            if db_upper in excluded_upper or db_upper.startswith(excluded_prefixes_upper):
                skipped_excluded_db.append((db_required, sql))
                return
            if db_upper not in orchard_dbs:
                skipped_missing_db.append((db_required, sql))
                return
        # SHOW GRANTS can return the same logical grant multiple times when it
        # was made by different grantors (Snowflake tracks granted_by per row);
        # those collapse to the same SQL here.
        if sql in seen_statements:
            skipped_duplicate += 1
            return
        seen_statements.add(sql)
        pending.append((sort_key, sql))

    for row in role_grants:
        consider(render_role_grant(row), role_grant_key(row), role_grant_sort_key(row))
    for row in future_role_grants:
        consider(
            render_future_role_grant(row),
            future_role_grant_key(row),
            future_role_grant_sort_key(row),
        )
    for row in user_grants:
        consider(render_user_grant(row), user_grant_key(row), user_grant_sort_key(row))

    # Sort by (grant type, target object, grantee), then strip keys.
    pending.sort(key=lambda item: item[0])
    statements: list[str] = [sql for _, sql in pending]

    total_discovered = len(role_grants) + len(future_role_grants) + len(user_grants)

    # Phase 6 — apply or write
    applied = 0
    failures: list[tuple[str, str]] = []
    failures_written_to: str | None = None
    if args.output_file:
        write_to_file(args.output_file, statements, args.delphi_account, args.orchard_account)
    else:
        applied, failures = apply_to_orchard(orchard_params, statements)
        if failures:
            write_failures_to_file(
                args.failures_file, failures, args.delphi_account, args.orchard_account
            )
            failures_written_to = args.failures_file

    # Phase 7 — final report
    sep = "=" * 74
    print()
    print(sep)
    print(f"GRANT SYNC SUMMARY: {args.delphi_account} → {args.orchard_account}")
    print(sep)
    print(f"Delphi roles scanned:        {len(roles):,}")
    print(f"Delphi users scanned:        {len(users):,}")
    print(f"Total grants discovered:     {total_discovered:,}")
    print(f"  Role grants:               {len(role_grants):,}")
    print(f"  Future role grants:        {len(future_role_grants):,}")
    print(f"  User grants:               {len(user_grants):,}")
    if unrenderable:
        print(f"Unrenderable rows:           {unrenderable:,}")
    print(f"Skipped (already in Orchard):{skipped_already_in_orchard:>5,}")
    print(f"Skipped (duplicate row):     {skipped_duplicate:,}")
    print(f"Skipped (grantee excluded):  {skipped_excluded_grantee:,}")
    print(f"Skipped (granted role excl): {skipped_excluded_granted_role:,}")
    print(f"Skipped (object type excl):  {skipped_excluded_object_type:,}")
    print(f"Skipped (integration excl):  {skipped_excluded_integration:,}")
    print(f"Skipped (ownership):         {skipped_ownership:,}")
    print(f"Skipped (db not in Orchard): {len(skipped_missing_db):,}")
    if skipped_missing_db:
        missing_dbs = sorted({db.upper() for db, _ in skipped_missing_db})
        print(f"  Missing databases ({len(missing_dbs)}): {', '.join(missing_dbs)}")
    print(f"Skipped (db excluded):       {len(skipped_excluded_db):,}")
    if skipped_excluded_db:
        excluded_dbs = sorted({db.upper() for db, _ in skipped_excluded_db})
        print(f"  Excluded databases ({len(excluded_dbs)}): {', '.join(excluded_dbs)}")
    print(f"Statements emitted:          {len(statements):,}")
    if args.output_file:
        print(f"Written to:                  {args.output_file}")
    else:
        print(f"  Applied successfully:      {applied:,}")
        print(f"  Failed:                    {len(failures):,}")
        if failures_written_to:
            print(f"  Failed statements written to: {failures_written_to}")

    if failures:
        print()
        print(f"FAILURES ({len(failures)}):")
        for stmt, err in failures:
            print(f"  FAILED: {stmt}")
            print(f"    └─ {err}")

    print(sep)


if __name__ == "__main__":
    main()