"""
Snowflake connection. In Lambda, reads the PEM private key from config
(loaded at cold start from Secrets Manager). Falls back to
SNOWFLAKE_PRIVATE_KEY_PATH for local development.
"""

import logging
import os
import re
from contextlib import contextmanager
from pathlib import Path

import snowflake.connector
from cryptography.hazmat.primitives import serialization

import config

logger = logging.getLogger(__name__)

ONBOARDING_STATE_TABLE = f"{config.SNOWFLAKE_DATABASE}.{config.SNOWFLAKE_SCHEMA}.D2C_ONBOARDING_STATE"
REGISTRY_TABLE = f"{config.SNOWFLAKE_DATABASE}.{config.SNOWFLAKE_SCHEMA}.ECOMMERCE_STORE_REGISTRY"
SHOPIFY_DATABASE = config.SNOWFLAKE_DATABASE

# Hard domain → GP overrides for stores that can't be matched by name.
# Table lives in an env-specific operations schema (configured via SNOWFLAKE_DATABASE/SNOWFLAKE_SCHEMA).
# Default config uses QA_D2C_OPERATIONS; set SNOWFLAKE_SCHEMA=D2C_OPERATIONS in PROD.
# Examples:
#   QA:   SHOPIFY_STORES_GLOBAL.QA_D2C_OPERATIONS.D2C_GP_DOMAIN_OVERRIDES
#   PROD: SHOPIFY_STORES_GLOBAL.D2C_OPERATIONS.D2C_GP_DOMAIN_OVERRIDES
_GP_DOMAIN_OVERRIDES_TABLE = f"{config.SNOWFLAKE_DATABASE}.{config.SNOWFLAKE_SCHEMA}.D2C_GP_DOMAIN_OVERRIDES"

# Default guard against runaway statements. Kept tight because the vast majority
# of statements are simple upserts.
DEFAULT_STATEMENT_TIMEOUT_SECONDS = 30
# The GP/vendor resolution CTEs are heavier (leading-wildcard name matching over
# GLOBAL_PARTICIPANT), so they get more headroom to cut down false-null results
# caused purely by the tight default timeout.
RESOLUTION_STATEMENT_TIMEOUT_SECONDS = 60


def _sql_norm(col: str) -> str:
    return (
        f"lower(regexp_replace(regexp_replace({col}, '^\\\\s+|\\\\s+$', ''), "
        f"'^https?://(www\\\\.)?|^www\\\\.|/+$', ''))"
    )


def _load_private_key() -> bytes:
    if config.SNOWFLAKE_PRIVATE_KEY_PEM:
        pem_data = config.SNOWFLAKE_PRIVATE_KEY_PEM.encode()
    else:
        key_path = os.environ.get("SNOWFLAKE_PRIVATE_KEY_PATH", str(Path.home() / ".ssh/snowflake/rsa_key.p8"))
        with open(key_path, "rb") as f:
            pem_data = f.read()

    passphrase_raw = os.environ.get("SNOWFLAKE_KEY_PASSPHRASE", "").strip("'\"") or None
    password = passphrase_raw.encode() if passphrase_raw else None

    try:
        p_key = serialization.load_pem_private_key(pem_data, password=password)
    except TypeError:
        p_key = serialization.load_pem_private_key(pem_data, password=None)

    return p_key.private_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )


def get_connection() -> snowflake.connector.SnowflakeConnection:
    return snowflake.connector.connect(
        account=config.SNOWFLAKE_ACCOUNT,
        user=config.SNOWFLAKE_USER,
        role=config.SNOWFLAKE_ROLE,
        warehouse=config.SNOWFLAKE_WAREHOUSE,
        database=config.SNOWFLAKE_DATABASE,
        schema=config.SNOWFLAKE_SCHEMA,
        private_key=_load_private_key(),
        session_parameters={
            "QUERY_TAG": "shopify-data-connector-onboarding",
            "STATEMENT_TIMEOUT_IN_SECONDS": DEFAULT_STATEMENT_TIMEOUT_SECONDS,
        },
    )


@contextmanager
def cursor(conn: snowflake.connector.SnowflakeConnection):
    cur = conn.cursor()
    try:
        yield cur
    finally:
        cur.close()


@contextmanager
def _statement_timeout(conn: snowflake.connector.SnowflakeConnection, seconds: int):
    """Temporarily raise STATEMENT_TIMEOUT_IN_SECONDS, restoring the default on exit.

    The session default is deliberately tight; heavier resolution queries need
    more room. Always resets to DEFAULT_STATEMENT_TIMEOUT_SECONDS afterwards so
    the extra headroom never leaks to unrelated statements on the connection.
    """
    with cursor(conn) as cur:
        cur.execute(f"ALTER SESSION SET STATEMENT_TIMEOUT_IN_SECONDS = {int(seconds)}")
    try:
        yield
    finally:
        with cursor(conn) as cur:
            cur.execute(f"ALTER SESSION SET STATEMENT_TIMEOUT_IN_SECONDS = {DEFAULT_STATEMENT_TIMEOUT_SECONDS}")


def upsert_state(conn: snowflake.connector.SnowflakeConnection, data: dict) -> None:
    with cursor(conn) as cur:
        cur.execute(
            f"""
            MERGE INTO {ONBOARDING_STATE_TABLE} AS target
            USING (
                SELECT
                    %(shop_domain)s          AS shop_domain,
                    %(schema_name)s          AS schema_name,
                    %(connector_id)s         AS connector_id,
                    %(status)s               AS status,
                    %(error_message)s        AS error_message,
                    %(connect_card_uri)s     AS connect_card_uri,
                    %(link_created_at)s      AS link_created_at,
                    %(link_expires_at)s      AS link_expires_at,
                    %(connected_at)s         AS connected_at,
                    %(merch_company)s        AS merch_company,
                    %(selling_country)s      AS selling_country,
                    %(alt_myshopify_domain)s AS alt_myshopify_domain,
                    %(custom_domain)s        AS custom_domain,
                    CURRENT_TIMESTAMP()      AS updated_at
            ) AS staged ON LOWER(target.shop_domain) = LOWER(staged.shop_domain)
            WHEN MATCHED THEN UPDATE SET
                schema_name          = COALESCE(staged.schema_name, target.schema_name),
                connector_id         = COALESCE(staged.connector_id, target.connector_id),
                status               = COALESCE(staged.status, target.status),
                error_message        = staged.error_message,
                connect_card_uri     = staged.connect_card_uri,
                link_created_at      = staged.link_created_at,
                link_expires_at      = staged.link_expires_at,
                connected_at         = staged.connected_at,
                merch_company        = COALESCE(staged.merch_company, target.merch_company),
                selling_country      = COALESCE(staged.selling_country, target.selling_country),
                alt_myshopify_domain = COALESCE(staged.alt_myshopify_domain, target.alt_myshopify_domain),
                custom_domain        = COALESCE(staged.custom_domain, target.custom_domain),
                updated_at           = staged.updated_at
            WHEN NOT MATCHED THEN INSERT (
                shop_domain, schema_name, connector_id, status,
                error_message, connect_card_uri, link_created_at, link_expires_at,
                connected_at, merch_company, selling_country, alt_myshopify_domain,
                custom_domain, updated_at
            ) VALUES (
                staged.shop_domain, staged.schema_name, staged.connector_id, staged.status,
                staged.error_message, staged.connect_card_uri, staged.link_created_at, staged.link_expires_at,
                staged.connected_at, staged.merch_company, staged.selling_country, staged.alt_myshopify_domain,
                staged.custom_domain, staged.updated_at
            )
        """,
            {
                k: data.get(k)
                for k in [
                    "shop_domain",
                    "schema_name",
                    "connector_id",
                    "status",
                    "error_message",
                    "connect_card_uri",
                    "link_created_at",
                    "link_expires_at",
                    "connected_at",
                    "merch_company",
                    "selling_country",
                    "alt_myshopify_domain",
                    "custom_domain",
                ]
            },
        )


def get_stores_by_statuses(conn: snowflake.connector.SnowflakeConnection, statuses: list[str]) -> list[dict]:
    placeholders = ", ".join(f"'{s}'" for s in statuses)
    with cursor(conn) as cur:
        cur.execute(f"""
            SELECT shop_domain, schema_name, connector_id,
                   status, error_message, connect_card_uri, link_expires_at, merch_company,
                   selling_country, rep_owner, alt_myshopify_domain, custom_domain,
                   connected_at, tables_synced, sync_frequency
            FROM {ONBOARDING_STATE_TABLE}
            WHERE status IN ({placeholders})
            ORDER BY shop_domain
        """)
        cols = [d[0].lower() for d in cur.description]
        return [dict(zip(cols, row)) for row in cur.fetchall()]


def set_status_by_connector_id(
    conn: snowflake.connector.SnowflakeConnection,
    connector_id: str,
    status: str,
    error_message: str | None = None,
    connected_at: str | None = None,
    tables_synced: int | None = None,
    sync_frequency: int | None = None,
) -> None:
    with cursor(conn) as cur:
        cur.execute(
            f"""
            UPDATE {ONBOARDING_STATE_TABLE}
            SET status = %(status)s,
                error_message = %(error_message)s,
                connected_at = COALESCE(%(connected_at)s, connected_at),
                tables_synced = COALESCE(%(tables_synced)s, tables_synced),
                sync_frequency = COALESCE(%(sync_frequency)s, sync_frequency),
                updated_at = CURRENT_TIMESTAMP()
            WHERE connector_id = %(connector_id)s
        """,
            {
                "connector_id": connector_id,
                "status": status,
                "error_message": error_message,
                "connected_at": connected_at,
                "tables_synced": tables_synced,
                "sync_frequency": sync_frequency,
            },
        )


def shop_table_exists(conn: snowflake.connector.SnowflakeConnection, schema_name: str) -> bool:
    with cursor(conn) as cur:
        cur.execute(
            f"""
            SELECT 1
            FROM {SHOPIFY_DATABASE}.INFORMATION_SCHEMA.TABLES
            WHERE table_schema = UPPER(%(schema_name)s)
              AND table_name   = 'SHOP'
              AND table_type   = 'BASE TABLE'
            LIMIT 1
        """,
            {"schema_name": schema_name},
        )
        return cur.fetchone() is not None


# Merch companies where GP → product → release resolution is attempted,
# matching the logic in the registry create script exactly.
_GP_RESOLVABLE_MERCH_COMPANIES = {
    "Ceremony of Roses",
    "CM Distro",
    "30° Merchandising",
    "SME France",
    "Other",
}

# Fallback vendor IDs when GP resolution finds nothing, keyed by merch company.
_MERCH_COMPANY_FALLBACK_VENDORS = {
    "Ceremony of Roses": 11111,
    "CM Distro": 34562,
    "30° Merchandising": 34562,
    "SME France": 34562,
    "Other": 34562,
}


def _validate_schema_name(schema_name: str | None) -> None:
    # Match Snowflake's unquoted-identifier rules: start with a letter or
    # underscore, then letters/digits/underscore/$. A digit-leading name passes
    # a naive [A-Za-z0-9_]+ check but is not a valid unquoted identifier, so the
    # f-string interpolation in get_store_info() would fail at runtime.
    if not schema_name or not re.match(r"^[A-Za-z_][A-Za-z0-9_$]*$", schema_name):
        raise ValueError(f"Invalid schema_name for SQL identifier: {schema_name!r}")


def get_store_info(
    conn: snowflake.connector.SnowflakeConnection,
    schema_name: str | None,
) -> tuple[str | None, int | None, str | None, str | None]:
    """Return (store_name, shopify_store_id, country_code, currency) from the SHOP table."""
    _validate_schema_name(schema_name)
    with cursor(conn) as cur:
        cur.execute(f"""
            SELECT NAME, ID, COUNTRY_CODE, CURRENCY
            FROM {SHOPIFY_DATABASE}.{schema_name}.SHOP
            WHERE _FIVETRAN_DELETED = FALSE
            LIMIT 1
        """)
        row = cur.fetchone()
    if not row:
        return None, None, None, None
    name = (row[0] or "").strip() or None
    return name, int(row[1]) if row[1] is not None else None, row[2], row[3]


def resolve_vendor_id(
    conn: snowflake.connector.SnowflakeConnection,
    gp_id: str | None,
    merch_company: str | None,
    shop_domain: str | None = None,
) -> tuple[int | None, str | None, str | None, str | None]:
    """
    Resolves vendor fields for a store at sync_complete time.

    Returns (vendor_id, vendor_name, vendor_brand, resolution_method).
    vendor_brand is always None until FACTS.PROD.VENDOR_COMPANY_BRAND_PARENT_COMPANY
    access is granted to the Lambda role.

    Resolution order:
      Kings Road  → 20977 (SOURCE_RULE, no DB interaction)
      IndieMerch  → 791836 (SOURCE_RULE, no DB interaction)
      GP_PRODUCT  → highest-product-count vendor linked to the GP
      ARTIST_STORES → ARTIST_STORES.MAPPED_LABEL_ID
      SOURCE_RULE → merch-company fallback vendor
      UNRESOLVED  → (None, None, None, None)
    """
    if merch_company in {"Kings Road", "Kings Road Merch EU", "Kings Road Merch US"}:
        return 20977, None, None, "SOURCE_RULE"
    if merch_company == "IndieMerch":
        return 791836, None, None, "SOURCE_RULE"
    if merch_company not in _GP_RESOLVABLE_MERCH_COMPANIES:
        return None, None, None, None

    if gp_id:
        with _statement_timeout(conn, RESOLUTION_STATEMENT_TIMEOUT_SECONDS), cursor(conn) as cur:
            cur.execute(
                """
                SELECT dr.LABELID AS vendor_id, MIN(v.NAME) AS vendor_name,
                       COUNT(DISTINCT dr.DISPLAY_UPC) AS cnt
                FROM FACTS.PROD.GLOBAL_PARTICIPANT_BY_PRODUCT_ID_BY_ISRC gpi
                JOIN FACTS.PROD.DIM_RELEASE dr
                    ON dr.PRODUCT_ID = gpi.PRODUCT_ID
                JOIN ORCHARD_APP_REPORTING.DELPHI_PROD.VENDOR v
                    ON v.VENDOR_ID = dr.LABELID
                   AND (v.LABEL_IDENTIFIER = 'D3' OR v.LABEL_IDENTIFIER IS NULL)
                   AND v.STATUS = 'signed'
                   AND v._FIVETRAN_DELETED = FALSE
                WHERE gpi.ID = %(gp_id)s
                GROUP BY dr.LABELID
                ORDER BY cnt DESC
                LIMIT 1
            """,
                {"gp_id": gp_id},
            )
            row = cur.fetchone()

        if row and row[0]:
            return int(row[0]), row[1] or None, None, "GP_PRODUCT"

    # GP resolution found nothing — check ARTIST_STORES before using hardcoded fallback
    if shop_domain:
        with cursor(conn) as cur:
            cur.execute(
                """
                SELECT TRY_CAST(MAPPED_LABEL_ID AS NUMBER)
                FROM ORCHARD_APP_REPORTING_V2.D2C.ARTIST_STORES
                WHERE lower(regexp_replace(
                          split_part(split_part(split_part(
                              regexp_replace(trim(SHOPIFY_STORE_URL), '^https?://', '', 1, 0, 'i'),
                          '/', 1), '?', 1), '#', 1),
                      '^www\\.', '', 1, 0, 'i')) = lower(%(domain)s)
                  AND MAPPED_LABEL_ID IS NOT NULL
                LIMIT 1
            """,
                {"domain": shop_domain},
            )
            row = cur.fetchone()

        if row and row[0]:
            return int(row[0]), None, None, "ARTIST_STORES"

    fallback = _MERCH_COMPANY_FALLBACK_VENDORS.get(merch_company)
    if fallback is not None:
        return fallback, None, None, "SOURCE_RULE"
    return None, None, None, None


def resolve_gp(
    conn: snowflake.connector.SnowflakeConnection,
    store_name: str | None,
    shop_domain: str | None = None,
) -> tuple[str | None, str | None]:
    """
    Resolves GLOBAL_PARTICIPANT_ID and name for a store. Returns (None, None) if
    no match.

    Resolution order:
      1. Domain override — check D2C_GP_DOMAIN_OVERRIDES for the shop domain.
         Short-circuits name-matching when found.
      2. Name-matching against GLOBAL_PARTICIPANT, with match priorities:
           1  — exact name match on store_name
           2  — exact match after stripping regional/format suffixes (cleaned)
           3  — exact match after double suffix strip (double_cleaned)
           4  — exact match after dash-suffix strip (triple_cleaned)
           5-7 — NORMALIZED_NAME equivalents of cleaned / double / triple
           8-9 — raw NAME special-char-stripped vs store_name / triple_cleaned
           10  — partial: triple_cleaned is a leading or trailing whole word of
                 gp.NAME (requires LEN >= 5)

    Tie-breaking: highest product count wins, then lowest gp.ID.
    GPs with no products are included if they represent a public participant.
    If the matched GP has been merged, returns the canonical post-merge ID/name.
    """
    if shop_domain:
        norm = re.sub(r"^https?://(www\.)?|^www\.", "", shop_domain, flags=re.IGNORECASE).rstrip("/")
        try:
            with cursor(conn) as cur:
                cur.execute(
                    f"""
                    SELECT
                        COALESCE(m.GLOBAL_PARTICIPANT_ID_TO, ovr.GLOBAL_PARTICIPANT_ID) AS gp_id,
                        COALESCE(merged.NAME, canonical.NAME)                           AS gp_name
                    FROM {_GP_DOMAIN_OVERRIDES_TABLE} ovr
                    JOIN FACTS.PROD.GLOBAL_PARTICIPANT canonical
                        ON canonical.ID = ovr.GLOBAL_PARTICIPANT_ID
                    LEFT JOIN FACTS.PROD.GLOBAL_PARTICIPANT_MERGED_TO_GLOBAL_PARTICIPANT m
                        ON m.GLOBAL_PARTICIPANT_ID_FROM = ovr.GLOBAL_PARTICIPANT_ID
                    LEFT JOIN FACTS.PROD.GLOBAL_PARTICIPANT merged
                        ON merged.ID = m.GLOBAL_PARTICIPANT_ID_TO
                    WHERE LOWER(ovr.SHOP_DOMAIN) = LOWER(%(domain)s)
                """,
                    {"domain": norm},
                )
                row = cur.fetchone()
            if row and row[0]:
                return row[0], row[1]
        except Exception as exc:
            logger.warning(
                f"[{shop_domain}] GP domain override lookup failed, falling through to name-matching: {exc}",
                exc_info=True,
            )

    if not store_name:
        return None, None

    with _statement_timeout(conn, RESOLUTION_STATEMENT_TIMEOUT_SECONDS), cursor(conn) as cur:
        cur.execute(
            """
            WITH input AS (
                SELECT
                    %(n)s AS store_name,
                    REGEXP_REPLACE(
                        %(n)s,
                        '\\\\s+(UK|US|EU|AU|ANZ|FR|DE|JP|SEA|Asia|APAC|Official Store|Store|'
                        || 'Merch|Shop|Official|US Store|UK Store|EU Store|EU/UK Store|EU/UK|'
                        || 'Preview|ANZ Store)$',
                        '', 1, 0, 'i') AS cleaned,
                    REGEXP_REPLACE(
                        REGEXP_REPLACE(%(n)s,
                            '\\\\s+(UK|US|EU|AU|ANZ|FR|DE|JP|SEA|Asia|APAC|EU/UK|EU/UK Store|ANZ Store)$',
                            '', 1, 0, 'i'),
                        '\\\\s+(Official Store|Store|Merch|Shop|Official|Music|Clothing|Records|Poppstar)$',
                        '', 1, 0, 'i') AS double_cleaned,
                    REGEXP_REPLACE(
                        REGEXP_REPLACE(
                            REGEXP_REPLACE(%(n)s,
                                '\\\\s+(UK|US|EU|AU|ANZ|FR|DE|JP|SEA|Asia|APAC|EU/UK|EU/UK Store|ANZ Store)$',
                                '', 1, 0, 'i'),
                            '\\\\s+(Official Store|Store|Merch|Shop|Official|Music|Clothing|Records|Poppstar)$',
                            '', 1, 0, 'i'),
                        '\\\\s*[-–—]\\\\s+.*$', '', 1, 0, 'i') AS triple_cleaned
            ),
            name_matches AS (
                SELECT
                    COALESCE(m.GLOBAL_PARTICIPANT_ID_TO, gp.ID) AS gp_id,
                    gp.ID                                       AS raw_gp_id,
                    CASE
                        WHEN LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.store_name))     THEN 1
                        WHEN LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.cleaned))        THEN 2
                        WHEN LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.double_cleaned)) THEN 3
                        WHEN LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.triple_cleaned)) THEN 4
                        WHEN LOWER(TRIM(gp.NORMALIZED_NAME)) = LOWER(
                            REGEXP_REPLACE(i.cleaned, '[^a-zA-Z0-9 ]', '')
                        ) THEN 5
                        WHEN LOWER(TRIM(gp.NORMALIZED_NAME)) = LOWER(
                            REGEXP_REPLACE(i.double_cleaned, '[^a-zA-Z0-9 ]', '')
                        ) THEN 6
                        WHEN LOWER(TRIM(gp.NORMALIZED_NAME)) = LOWER(
                            REGEXP_REPLACE(i.triple_cleaned, '[^a-zA-Z0-9 ]', '')
                        ) THEN 7
                        WHEN LOWER(REGEXP_REPLACE(gp.NAME, '[^a-zA-Z0-9 ]', '')) = LOWER(
                            REGEXP_REPLACE(i.store_name, '[^a-zA-Z0-9 ]', '')
                        ) THEN 8
                        WHEN LOWER(REGEXP_REPLACE(gp.NAME, '[^a-zA-Z0-9 ]', '')) = LOWER(
                            REGEXP_REPLACE(i.triple_cleaned, '[^a-zA-Z0-9 ]', '')
                        ) THEN 9
                        WHEN (LOWER(TRIM(gp.NAME)) LIKE '%% ' || LOWER(TRIM(i.triple_cleaned))
                           OR LOWER(TRIM(gp.NAME)) LIKE LOWER(TRIM(i.triple_cleaned)) || ' %%')
                          AND LEN(i.triple_cleaned) >= 5                           THEN 10
                    END AS match_priority
                FROM input i
                JOIN FACTS.PROD.GLOBAL_PARTICIPANT gp
                    ON LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.store_name))
                    OR LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.cleaned))
                    OR LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.double_cleaned))
                    OR LOWER(TRIM(gp.NAME)) = LOWER(TRIM(i.triple_cleaned))
                    OR LOWER(TRIM(gp.NORMALIZED_NAME)) = LOWER(REGEXP_REPLACE(i.cleaned,       '[^a-zA-Z0-9 ]', ''))
                    OR LOWER(TRIM(gp.NORMALIZED_NAME)) = LOWER(REGEXP_REPLACE(i.double_cleaned,'[^a-zA-Z0-9 ]', ''))
                    OR LOWER(TRIM(gp.NORMALIZED_NAME)) = LOWER(REGEXP_REPLACE(i.triple_cleaned,'[^a-zA-Z0-9 ]', ''))
                    OR LOWER(REGEXP_REPLACE(gp.NAME, '[^a-zA-Z0-9 ]', ''))
                        = LOWER(REGEXP_REPLACE(i.store_name, '[^a-zA-Z0-9 ]', ''))
                    OR LOWER(REGEXP_REPLACE(gp.NAME, '[^a-zA-Z0-9 ]', ''))
                        = LOWER(REGEXP_REPLACE(i.triple_cleaned, '[^a-zA-Z0-9 ]', ''))
                    OR (LOWER(TRIM(gp.NAME)) LIKE '%% ' || LOWER(TRIM(i.triple_cleaned))
                        AND LEN(i.triple_cleaned) >= 5)
                    OR (LOWER(TRIM(gp.NAME)) LIKE LOWER(TRIM(i.triple_cleaned)) || ' %%'
                        AND LEN(i.triple_cleaned) >= 5)
                LEFT JOIN FACTS.PROD.GLOBAL_PARTICIPANT_MERGED_TO_GLOBAL_PARTICIPANT m
                    ON m.GLOBAL_PARTICIPANT_ID_FROM = gp.ID
            ),
            gp_product_counts AS (
                -- Scoped to name-matched GPs only: a full-table GROUP BY dominated
                -- runtime; only the matched GPs' counts are ever used here.
                SELECT ID, COUNT(DISTINCT PRODUCT_ID) AS product_count
                FROM FACTS.PROD.GLOBAL_PARTICIPANT_BY_PRODUCT_ID_BY_ISRC
                WHERE ID IN (SELECT gp_id FROM name_matches)
                GROUP BY ID
            ),
            gp_has_public_participant AS (
                -- New GPs with no catalogue yet but who represent a known public
                -- participant should still match rather than return NULL.
                SELECT DISTINCT GLOBAL_PARTICIPANT_ID AS ID
                FROM FACTS.PROD.GLOBAL_PARTICIPANT_REPRESENTS_PUBLIC_PARTICIPANT
                WHERE GLOBAL_PARTICIPANT_ID IN (SELECT gp_id FROM name_matches)
            ),
            gp_match AS (
                SELECT
                    nm.gp_id,
                    COALESCE(pc.product_count, 0) AS product_count
                FROM name_matches nm
                LEFT JOIN gp_product_counts pc ON pc.ID = nm.gp_id
                LEFT JOIN gp_has_public_participant hpp ON hpp.ID = nm.gp_id
                WHERE COALESCE(pc.product_count, 0) > 0 OR hpp.ID IS NOT NULL
                QUALIFY ROW_NUMBER() OVER (
                    ORDER BY nm.match_priority ASC, product_count DESC, nm.raw_gp_id
                ) = 1
            )
            SELECT gm.gp_id, canonical.NAME AS gp_name
            FROM gp_match gm
            JOIN FACTS.PROD.GLOBAL_PARTICIPANT canonical ON canonical.ID = gm.gp_id
        """,
            {"n": store_name},
        )
        result = cur.fetchone()

    if result and result[0]:
        return result[0], result[1]
    return None, None


def upsert_registry(conn: snowflake.connector.SnowflakeConnection, data: dict) -> None:
    with cursor(conn) as cur:
        cur.execute(
            f"""
            MERGE INTO {REGISTRY_TABLE} AS target
            USING (
                SELECT
                    %(shopify_store_id)s          AS shopify_store_id,
                    %(myshopify_domain)s          AS myshopify_domain,
                    %(schema_name)s               AS schema_name,
                    %(store_name)s                AS store_name,
                    %(country_code)s              AS country_code,
                    %(currency)s                  AS currency,
                    %(source)s                    AS source,
                    %(connector_id)s              AS connector_id,
                    %(connected_at)s              AS connected_at,
                    %(selling_country)s           AS selling_country,
                    %(merch_company)s             AS merch_company,
                    %(rep_owner)s                 AS rep_owner,
                    %(alt_myshopify_domain)s      AS alt_myshopify_domain,
                    %(custom_domain)s             AS custom_domain,
                    %(sync_frequency)s            AS sync_frequency,
                    %(tables_synced)s             AS tables_synced,
                    %(vendor_id)s                 AS vendor_id,
                    %(vendor_name)s               AS vendor_name,
                    %(vendor_brand)s              AS vendor_brand,
                    %(vendor_resolution_method)s  AS vendor_resolution_method,
                    %(global_participant_id)s     AS global_participant_id,
                    %(global_participant_name)s   AS global_participant_name,
                    %(is_synced_to_snowflake)s    AS is_synced_to_snowflake
            ) AS staged ON {_sql_norm("target.myshopify_domain")} = {_sql_norm("staged.myshopify_domain")}
            WHEN MATCHED THEN UPDATE SET
                shopify_store_id        = COALESCE(staged.shopify_store_id, target.shopify_store_id),
                schema_name             = COALESCE(staged.schema_name, target.schema_name),
                store_name              = COALESCE(staged.store_name, target.store_name),
                country_code            = COALESCE(staged.country_code, target.country_code),
                currency                = COALESCE(staged.currency, target.currency),
                source                  = COALESCE(staged.source, target.source),
                connector_id            = COALESCE(staged.connector_id, target.connector_id),
                connected_at            = COALESCE(staged.connected_at, target.connected_at),
                selling_country         = COALESCE(staged.selling_country, target.selling_country),
                merch_company           = COALESCE(staged.merch_company, target.merch_company),
                rep_owner               = COALESCE(target.rep_owner, staged.rep_owner),
                alt_myshopify_domain    = COALESCE(staged.alt_myshopify_domain, target.alt_myshopify_domain),
                custom_domain           = COALESCE(staged.custom_domain, target.custom_domain),
                sync_frequency          = COALESCE(staged.sync_frequency, target.sync_frequency),
                tables_synced           = COALESCE(staged.tables_synced, target.tables_synced),
                vendor_id               = COALESCE(staged.vendor_id, target.vendor_id),
                vendor_name             = COALESCE(staged.vendor_name, target.vendor_name),
                vendor_brand            = COALESCE(staged.vendor_brand, target.vendor_brand),
                vendor_resolution_method = COALESCE(staged.vendor_resolution_method, target.vendor_resolution_method),
                global_participant_id   = COALESCE(staged.global_participant_id, target.global_participant_id),
                global_participant_name = COALESCE(staged.global_participant_name, target.global_participant_name),
                is_synced_to_snowflake  = COALESCE(staged.is_synced_to_snowflake, target.is_synced_to_snowflake)
            WHEN NOT MATCHED THEN INSERT (
                shopify_store_id, myshopify_domain, schema_name, store_name, country_code, currency,
                source, connector_id, connected_at, selling_country, merch_company, rep_owner,
                alt_myshopify_domain, custom_domain, sync_frequency, tables_synced,
                vendor_id, vendor_name, vendor_brand, vendor_resolution_method,
                global_participant_id, global_participant_name, is_synced_to_snowflake
            ) VALUES (
                staged.shopify_store_id, staged.myshopify_domain, staged.schema_name,
                staged.store_name, staged.country_code, staged.currency,
                staged.source, staged.connector_id, staged.connected_at,
                staged.selling_country, staged.merch_company, staged.rep_owner,
                staged.alt_myshopify_domain, staged.custom_domain,
                staged.sync_frequency, staged.tables_synced,
                staged.vendor_id, staged.vendor_name, staged.vendor_brand, staged.vendor_resolution_method,
                staged.global_participant_id, staged.global_participant_name,
                staged.is_synced_to_snowflake
            )
        """,
            {
                k: data.get(k)
                for k in [
                    "shopify_store_id",
                    "myshopify_domain",
                    "schema_name",
                    "store_name",
                    "country_code",
                    "currency",
                    "source",
                    "connector_id",
                    "connected_at",
                    "selling_country",
                    "merch_company",
                    "rep_owner",
                    "alt_myshopify_domain",
                    "custom_domain",
                    "sync_frequency",
                    "tables_synced",
                    "vendor_id",
                    "vendor_name",
                    "vendor_brand",
                    "vendor_resolution_method",
                    "global_participant_id",
                    "global_participant_name",
                    "is_synced_to_snowflake",
                ]
            },
        )


def set_synced_to_snowflake(
    conn: snowflake.connector.SnowflakeConnection,
    shop_domain: str,
    synced: bool,
) -> None:
    with cursor(conn) as cur:
        cur.execute(
            f"""
            UPDATE {REGISTRY_TABLE}
            SET IS_SYNCED_TO_SNOWFLAKE = %(synced)s
            WHERE LOWER(MYSHOPIFY_DOMAIN) = LOWER(%(shop_domain)s)
        """,
            {"shop_domain": shop_domain, "synced": synced},
        )