"""Push master taxonomy hierarchy to Magento categories and Shopify collections.

Local ``master_taxonomy_node`` is the source of truth for hierarchy and display names.
Magento receives categories (with parent IDs); Shopify receives flat collections
(``remote_parent_id`` is always ``0``).
"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Sequence

from sqlalchemy import inspect, select, update
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.catalog_intent_planner import ALL_PRODUCTS
from db.channel_listing_path import (
    ACTIVE_STATUS,
    INACTIVE_STATUS,
    bulk_assign_listing_paths,
    resolve_listing_paths_for_sku,
    upsert_listing_path,
    upsert_listing_path_target,
)
from db.master_taxonomy_hierarchy import (
    NODE_CATEGORY,
    NODE_COLLECTION,
    NODE_HUB,
    _match_magento_node,
    _magento_registry_rows,
    build_hierarchy_from_master_catalog,
    list_taxonomy_tree,
    sync_channel_links_from_registries,
)
from db.master_taxonomy_registry import backfill_master_taxonomy_from_products, registry_code_from_name
from db.magento_repositories import SqlAlchemyCategoryRegistryRepository
from db.models import (
    ChannelListingPath,
    ChannelListingPathTarget,
    CollectionLandingPage,
    CollectionSkuOverride,
    MagentoCategoryRegistry,
    MasterCollectionRegistry,
    MasterProduct,
    MasterTaxonomyChannelLink,
    MasterTaxonomyNode,
    ProductChannelTaxonomyAssignment,
    ProductListingPathAssignment,
    ShopifyCollectionRegistry,
)
from db.product_channel_taxonomy import (
    MAGENTO_KIND,
    SHOPIFY_COLLECTION_KIND,
    bulk_assign_sku_taxonomy,
    upsert_sku_taxonomy_assignment,
)

MAGENTO_ROOT_CATEGORY_ID = 2
MATCH_SOURCE = "master_taxonomy"


class TaxonomyChannelConflict(Exception):
    """Remote channel entity already exists and must be linked or renamed."""

    def __init__(
        self,
        *,
        channel: str,
        reason: str,
        handle: str,
        message: str,
        candidates: Optional[List[Dict[str, Any]]] = None,
        related_candidates: Optional[List[Dict[str, Any]]] = None,
        suggested_handles: Optional[List[str]] = None,
        path_slug: Optional[str] = None,
        guidance: Optional[Dict[str, Any]] = None,
    ):
        self.channel = channel
        self.reason = reason
        self.handle = handle
        self.candidates = list(candidates or [])
        self.related_candidates = list(related_candidates or [])
        self.suggested_handles = list(suggested_handles or [])
        self.path_slug = path_slug
        self.guidance = dict(guidance or {})
        super().__init__(message)


def bootstrap_master_taxonomy(
    session: Session,
    *,
    dry_run: bool = False,
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    run_backfill: bool = True,
    run_build_hierarchy: bool = True,
    run_sync_links: bool = True,
    push_to_channels: bool = False,
    magento_api: Optional[Any] = None,
    shopify_client: Optional[Any] = None,
    shopify_shop_code: Optional[str] = None,
    create_missing_remotes: bool = True,
    allow_invent: Optional[bool] = True,
) -> Dict[str, Any]:
    """One-shot bootstrap: backfill registries → build hierarchy → link pulled channel IDs.

    Dashboard/API intentional bootstrap defaults allow_invent=True.
    """
    from db.taxonomy_invent_policy import can_auto_invent

    invent_ok = can_auto_invent(allow_invent=allow_invent)
    result: Dict[str, Any] = {"dry_run": dry_run, "steps": {}, "invent_allowed": invent_ok}
    if run_backfill:
        result["steps"]["backfill"] = backfill_master_taxonomy_from_products(
            session, dry_run=dry_run, allow_invent=invent_ok
        )
    if run_build_hierarchy:
        result["steps"]["build_hierarchy"] = build_hierarchy_from_master_catalog(
            session, dry_run=dry_run, allow_invent=invent_ok
        )
    if run_sync_links:
        result["steps"]["sync_links"] = sync_channel_links_from_registries(
            session,
            magento_connection_id=magento_connection_id,
            shopify_connection_id=shopify_connection_id,
            dry_run=dry_run,
            write_listing_path_targets=True,
        )
    if push_to_channels and not dry_run:
        result["steps"]["push_channels"] = push_taxonomy_nodes_to_channels(
            session,
            magento_connection_id=magento_connection_id,
            shopify_connection_id=shopify_connection_id,
            magento_api=magento_api,
            shopify_client=shopify_client,
            shopify_shop_code=shopify_shop_code,
            dry_run=False,
            create_missing=create_missing_remotes,
        )
    result["tree"] = list_taxonomy_tree(session)
    return result


def upsert_taxonomy_node(
    session: Session,
    payload: Dict[str, Any],
    *,
    sync_listing_path: bool = True,
    allow_invent: Optional[bool] = True,
) -> Dict[str, Any]:
    """Create or update a master taxonomy node (local source of truth).

    Intentional dashboard/API upserts default allow_invent=True.
    Auto invent pipelines must pass allow_invent=False (or rely on proposal/hierarchy gates).
    """
    from db.taxonomy_invent_policy import can_auto_invent

    node_id = payload.get("id")
    name = str(payload.get("name") or "").strip()
    if not name:
        raise ValueError("name is required")

    parent_id = payload.get("parent_id")
    if parent_id in (0, "0", "", None):
        parent_id = None
    else:
        parent_id = int(parent_id)

    row = session.get(MasterTaxonomyNode, int(node_id)) if node_id else None
    if parent_id is None and row is not None and "parent_id" not in payload:
        parent_id = row.parent_id

    node_kind = str(
        payload.get("node_kind") or (row.node_kind if row else NODE_HUB)
    ).strip().lower()

    parent_changed = (
        row is not None
        and "parent_id" in payload
        and int(parent_id or 0) != int(row.parent_id or 0)
    )
    explicit_path = payload.get("path_slug")
    if explicit_path:
        path_slug = normalize_plp_path(str(explicit_path))
    elif row is not None and parent_changed:
        path_slug = _recompute_node_path_slug(session, row, parent_id)
    elif row is not None and row.path_slug:
        path_slug = normalize_plp_path(row.path_slug)
    else:
        path_slug = normalize_plp_path(_default_path_slug(session, name, parent_id))
    if not path_slug:
        raise ValueError("path_slug could not be resolved")

    # path_slug is the natural key. When callers omit id (proposal apply / pipeline),
    # reuse the existing row — including inactive nodes — instead of failing.
    if row is None:
        row = session.scalar(
            select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == path_slug).limit(1)
        )
        if row is not None:
            node_id = row.id
            if parent_id is None and "parent_id" not in payload:
                parent_id = row.parent_id
            if not payload.get("node_kind"):
                node_kind = str(row.node_kind or node_kind).strip().lower()
            parent_changed = (
                "parent_id" in payload and int(parent_id or 0) != int(row.parent_id or 0)
            )

    slug_check = select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == path_slug).limit(1)
    if node_id or (row is not None and row.id):
        slug_check = slug_check.where(MasterTaxonomyNode.id != int(node_id or row.id))
    existing_slug = session.scalar(slug_check)
    if existing_slug is not None:
        raise ValueError(f"path_slug already in use: {path_slug}")

    invent_ok = can_auto_invent(allow_invent=allow_invent)
    if row is None and not invent_ok:
        raise ValueError(
            f"Taxonomy invent is frozen; cannot create node for path_slug={path_slug!r}. "
            "Unlock invent or create the node via the dashboard API with invent enabled."
        )

    old_path_slug = row.path_slug if row else None
    old_name = row.name if row else None
    old_parent_id = row.parent_id if row else None
    collection_registry_id = payload.get("collection_registry_id")
    if collection_registry_id is None and row is not None:
        collection_registry_id = row.collection_registry_id
    if node_kind == NODE_COLLECTION and not collection_registry_id:
        collection_registry_id = _ensure_collection_registry(
            session, name, parent_id, allow_invent=invent_ok
        )

    data = {
        "parent_id": parent_id,
        "node_kind": node_kind,
        "code": str(payload.get("code") or registry_code_from_name(name)),
        "name": name,
        "path_slug": path_slug,
        "collection_registry_id": collection_registry_id,
        "sort_order": int(payload.get("sort_order") or (row.sort_order if row else 100)),
        "is_active": bool(payload.get("is_active", True)),
        "notes": payload.get("notes"),
    }
    created = row is None
    if row is None:
        row = MasterTaxonomyNode(**data)
        session.add(row)
    else:
        for key, value in data.items():
            if value is not None:
                setattr(row, key, value)
    if "node_kind" not in payload or payload.get("node_kind") in (None, ""):
        _sync_node_kind_for_parent(row)
    session.flush()

    cascade_result: Dict[str, Any] = {"updated_nodes": [], "cascade_count": 0}
    if old_path_slug and old_path_slug != row.path_slug:
        cascade_result = _apply_taxonomy_path_cascade(
            session,
            row,
            old_path_slug,
            row.path_slug,
            sync_listing_path=sync_listing_path,
        )
    elif sync_listing_path:
        upsert_listing_path(
            session,
            path_slug=row.path_slug,
            title=row.name,
            path_kind=row.node_kind,
            parent_path_slug=_parent_path_slug(session, row.parent_id),
            notes="master taxonomy node",
        )

    _apply_manual_channel_links(session, row, payload)

    if "category_icon_url" in payload and row.node_kind in {NODE_HUB, NODE_CATEGORY}:
        _upsert_taxonomy_category_icon(session, row, payload.get("category_icon_url"))

    landing_link = None
    if row.node_kind == NODE_COLLECTION:
        landing_link = ensure_collection_landing_link(session, row)
        if row.collection_registry_id:
            reg = session.get(MasterCollectionRegistry, int(row.collection_registry_id))
            if reg:
                reg.name = _leaf_title(row.name)
                reg.path_slug = row.path_slug

    links = _links_for_node(session, row.id)
    changed_remote_fields = {
        "name": old_name != row.name if old_name else True,
        "path_slug": old_path_slug != row.path_slug if old_path_slug else True,
        "parent_id": old_parent_id != row.parent_id if row else True,
    }
    icon_url = None
    if row.node_kind in {NODE_HUB, NODE_CATEGORY}:
        landing = session.scalar(
            select(CollectionLandingPage).where(CollectionLandingPage.path_slug == row.path_slug).limit(1)
        )
        icon_url = landing.category_icon_url if landing else None
    return {
        "created": created,
        "node": _node_payload(row, links, category_icon_url=icon_url),
        "landing_link": landing_link,
        "changed_remote_fields": changed_remote_fields,
        "path_cascade": cascade_result,
    }


def reparent_taxonomy_node(
    session: Session,
    node_id: int,
    *,
    parent_id: Optional[int] = None,
    sort_order: Optional[int] = None,
) -> Dict[str, Any]:
    """Move a node under a new parent (drag-and-drop). Collections keep registry link."""
    node = session.get(MasterTaxonomyNode, int(node_id))
    if node is None or not node.is_active:
        raise ValueError(f"Unknown taxonomy node: {node_id}")

    new_parent_id = None if parent_id in (None, 0, "0") else int(parent_id)
    if new_parent_id == node.id:
        raise ValueError("A node cannot be its own parent")
    if new_parent_id is not None:
        parent = session.get(MasterTaxonomyNode, new_parent_id)
        if parent is None or not parent.is_active:
            raise ValueError(f"Unknown parent node: {new_parent_id}")
        if _is_descendant(session, ancestor_id=node.id, node_id=new_parent_id):
            raise ValueError("Cannot move a node under its own descendant")

    old_path_slug = node.path_slug
    node.parent_id = new_parent_id
    _sync_node_kind_for_parent(node)
    if sort_order is not None:
        node.sort_order = int(sort_order)
    session.flush()

    new_path_slug = _recompute_node_path_slug(session, node, new_parent_id)
    cascade_result: Dict[str, Any] = {"updated_nodes": [], "cascade_count": 0}
    if old_path_slug != new_path_slug:
        cascade_result = _apply_taxonomy_path_cascade(
            session,
            node,
            old_path_slug,
            new_path_slug,
            sync_listing_path=True,
        )
    else:
        upsert_listing_path(
            session,
            path_slug=node.path_slug,
            title=node.name,
            path_kind=node.node_kind,
            parent_path_slug=_parent_path_slug(session, node.parent_id),
            notes="master taxonomy reparent",
        )
        if node.node_kind == NODE_COLLECTION:
            ensure_collection_landing_link(session, node)

    links = _links_for_node(session, node.id)
    return {
        "node": _node_payload(node, links),
        "changed_remote_fields": {"parent_id": True, "path_slug": old_path_slug != new_path_slug},
        "path_cascade": cascade_result,
    }


def _is_descendant(session: Session, *, ancestor_id: int, node_id: int) -> bool:
    current = session.get(MasterTaxonomyNode, int(node_id))
    seen: set[int] = set()
    while current and current.parent_id:
        if current.parent_id in seen:
            break
        if int(current.parent_id) == int(ancestor_id):
            return True
        seen.add(int(current.parent_id))
        current = session.get(MasterTaxonomyNode, int(current.parent_id))
    return False


def _apply_manual_channel_links(session: Session, node: MasterTaxonomyNode, payload: Dict[str, Any]) -> None:
    magento_id = str(payload.get("magento_remote_id") or "").strip()
    shopify_id = str(payload.get("shopify_remote_id") or "").strip()
    magento_conn = payload.get("magento_connection_id")
    shopify_conn = payload.get("shopify_connection_id")
    if magento_id and magento_conn is not None:
        kind = "collection" if node.node_kind == NODE_COLLECTION else MAGENTO_KIND
        _upsert_channel_link(
            session,
            node_id=node.id,
            channel_code="magento",
            connection_id=int(magento_conn),
            taxonomy_kind=kind,
            remote_id=magento_id,
            remote_parent_id=str(payload.get("magento_remote_parent_id") or MAGENTO_ROOT_CATEGORY_ID),
            remote_path=payload.get("magento_remote_path") or node.path_slug,
        )
        upsert_listing_path_target(
            session,
            path_slug=node.path_slug,
            channel_code="magento",
            remote_id=magento_id,
            taxonomy_kind=kind,
            connection_id=int(magento_conn),
            remote_path=payload.get("magento_remote_path") or node.path_slug,
        )
    if shopify_id and shopify_conn is not None:
        _upsert_channel_link(
            session,
            node_id=node.id,
            channel_code="shopify",
            connection_id=int(shopify_conn),
            taxonomy_kind=SHOPIFY_COLLECTION_KIND,
            remote_id=shopify_id,
            remote_parent_id="0",
            remote_path=payload.get("shopify_remote_path") or node.path_slug,
        )
        upsert_listing_path_target(
            session,
            path_slug=node.path_slug,
            channel_code="shopify",
            remote_id=shopify_id,
            taxonomy_kind=SHOPIFY_COLLECTION_KIND,
            connection_id=int(shopify_conn),
            remote_path=payload.get("shopify_remote_path") or node.path_slug,
        )
    session.flush()


def ensure_collection_landing_link(session: Session, node: MasterTaxonomyNode) -> Dict[str, Any]:
    """Ensure a collection taxonomy node has registry + landing page rows."""
    if node.node_kind != NODE_COLLECTION:
        return {"ensured": False, "reason": "not_a_collection"}

    from db.collection_landing_pages import DEFAULT_CATEGORY_L1, upsert_collection_landing_page
    from db.collection_taxonomy_bridge import link_collection_registry_to_landing

    category_l1 = DEFAULT_CATEGORY_L1
    if node.parent_id:
        parent = session.get(MasterTaxonomyNode, int(node.parent_id))
        if parent:
            category_l1 = _leaf_title(parent.name)

    parent_slug = _parent_path_slug(session, node.parent_id)
    payload: Dict[str, Any] = {
        "path_slug": node.path_slug,
        "title": node.name,
        "collection": _leaf_title(node.name),
        "category_l1": category_l1,
        "parent_path_slug": parent_slug,
    }
    if node.collection_registry_id:
        payload["collection_registry_id"] = int(node.collection_registry_id)

    landing = upsert_collection_landing_page(session, payload)
    bridge: Dict[str, Any] = {"linked": False}
    if node.collection_registry_id:
        try:
            bridge = link_collection_registry_to_landing(session, registry_id=int(node.collection_registry_id))
        except ValueError:
            bridge = link_collection_registry_to_landing(session, path_slug=node.path_slug)
    elif node.path_slug:
        try:
            bridge = link_collection_registry_to_landing(session, path_slug=node.path_slug)
        except ValueError:
            pass

    registry_id = (bridge.get("registry") or {}).get("id")
    if registry_id and not node.collection_registry_id:
        node.collection_registry_id = int(registry_id)
        session.flush()

    return {
        "ensured": True,
        "landing_page": landing,
        "registry_link": bridge,
    }


def _migrate_taxonomy_path_slug(
    session: Session,
    node: MasterTaxonomyNode,
    old_slug: str,
    new_slug: str,
) -> None:
    """Move local rows keyed by path_slug when a taxonomy node path changes."""
    old_slug = normalize_plp_path(old_slug)
    new_slug = normalize_plp_path(new_slug)
    if not old_slug or not new_slug or old_slug == new_slug:
        return

    listing = session.scalar(
        select(ChannelListingPath).where(ChannelListingPath.path_slug == old_slug).limit(1)
    )
    if listing is not None:
        listing.path_slug = new_slug
        if listing.parent_path_slug == old_slug:
            listing.parent_path_slug = new_slug

    landing = session.scalar(
        select(CollectionLandingPage).where(CollectionLandingPage.path_slug == old_slug).limit(1)
    )
    if landing is None and node.collection_registry_id:
        landing = session.scalar(
            select(CollectionLandingPage)
            .where(CollectionLandingPage.collection_registry_id == int(node.collection_registry_id))
            .limit(1)
        )
    if landing is not None:
        landing.path_slug = new_slug
        landing.title = node.name or landing.title
        landing.collection = _leaf_title(node.name) or landing.collection
        landing.parent_path_slug = _parent_path_slug(session, node.parent_id) or landing.parent_path_slug

    if inspect(session.bind).has_table(CollectionSkuOverride.__tablename__):
        for override in session.scalars(
            select(CollectionSkuOverride).where(CollectionSkuOverride.path_slug == old_slug)
        ).all():
            override.path_slug = new_slug

    if node.collection_registry_id:
        reg = session.get(MasterCollectionRegistry, int(node.collection_registry_id))
        if reg is not None:
            reg.path_slug = new_slug

    session.flush()


def _list_descendant_nodes(session: Session, root_id: int) -> List[MasterTaxonomyNode]:
    """Return all active descendants of a taxonomy node (depth-first order)."""
    out: List[MasterTaxonomyNode] = []
    stack = list(
        session.scalars(
            select(MasterTaxonomyNode)
            .where(MasterTaxonomyNode.parent_id == int(root_id))
            .where(MasterTaxonomyNode.is_active.is_(True))
            .order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name)
        ).all()
    )
    while stack:
        node = stack.pop(0)
        out.append(node)
        children = list(
            session.scalars(
                select(MasterTaxonomyNode)
                .where(MasterTaxonomyNode.parent_id == node.id)
                .where(MasterTaxonomyNode.is_active.is_(True))
                .order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name)
            ).all()
        )
        stack[:0] = children
    return out


def _path_leaf_segment(node: MasterTaxonomyNode) -> str:
    from db.catalog_intent_planner import slugify_segment

    slug = normalize_plp_path(node.path_slug or "")
    if slug:
        return slug.rsplit("/", 1)[-1]
    return slugify_segment(_leaf_title(node.name))


def _recompute_node_path_slug(
    session: Session,
    node: MasterTaxonomyNode,
    parent_id: Optional[int],
) -> str:
    leaf = _path_leaf_segment(node)
    if not parent_id:
        return normalize_plp_path(leaf)
    parent = session.get(MasterTaxonomyNode, int(parent_id))
    if parent and parent.path_slug:
        return normalize_plp_path(f"{parent.path_slug}/{leaf}")
    return normalize_plp_path(leaf)


def _plan_path_prefix_cascade(
    session: Session,
    root: MasterTaxonomyNode,
    old_prefix: str,
    new_prefix: str,
) -> List[Dict[str, Any]]:
    """Plan path_slug updates for root + descendants when a prefix changes."""
    old_prefix = normalize_plp_path(old_prefix)
    new_prefix = normalize_plp_path(new_prefix)
    if not old_prefix or old_prefix == new_prefix:
        return []

    planned: List[Dict[str, Any]] = []
    if old_prefix != new_prefix:
        planned.append(
            {
                "id": root.id,
                "node": root,
                "old_path_slug": old_prefix,
                "new_path_slug": new_prefix,
            }
        )

    for child in _list_descendant_nodes(session, root.id):
        child_old = normalize_plp_path(child.path_slug or "")
        if not child_old.startswith(old_prefix + "/"):
            continue
        child_new = normalize_plp_path(new_prefix + child_old[len(old_prefix) :])
        planned.append(
            {
                "id": child.id,
                "node": child,
                "old_path_slug": child_old,
                "new_path_slug": child_new,
            }
        )

    by_id: Dict[int, Dict[str, Any]] = {}
    for row in planned:
        by_id[int(row["id"])] = row
    return list(by_id.values())


def _assert_planned_path_slugs_available(session: Session, planned: List[Dict[str, Any]]) -> None:
    for row in planned:
        existing = session.scalar(
            select(MasterTaxonomyNode)
            .where(MasterTaxonomyNode.path_slug == row["new_path_slug"])
            .where(MasterTaxonomyNode.id != int(row["id"]))
            .limit(1)
        )
        if existing is not None:
            raise ValueError(
                f"path_slug already in use: {row['new_path_slug']} (conflicts with node {existing.id})"
            )


def _sync_taxonomy_node_local_paths(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    sync_listing_path: bool,
) -> None:
    if sync_listing_path:
        upsert_listing_path(
            session,
            path_slug=node.path_slug,
            title=node.name,
            path_kind=node.node_kind,
            parent_path_slug=_parent_path_slug(session, node.parent_id),
            notes="master taxonomy path cascade",
        )
    if node.node_kind == NODE_COLLECTION:
        landing = None
        if node.path_slug:
            landing = session.scalar(
                select(CollectionLandingPage).where(CollectionLandingPage.path_slug == node.path_slug).limit(1)
            )
        if landing is None and node.collection_registry_id:
            landing = session.scalar(
                select(CollectionLandingPage)
                .where(CollectionLandingPage.collection_registry_id == int(node.collection_registry_id))
                .limit(1)
            )
        if landing is None:
            pass
        else:
            landing.path_slug = node.path_slug
            landing.parent_path_slug = _parent_path_slug(session, node.parent_id)
            if node.collection_registry_id:
                landing.collection_registry_id = int(node.collection_registry_id)
        if node.collection_registry_id:
            reg = session.get(MasterCollectionRegistry, int(node.collection_registry_id))
            if reg is not None:
                reg.path_slug = node.path_slug


def _apply_taxonomy_path_cascade(
    session: Session,
    root: MasterTaxonomyNode,
    old_prefix: str,
    new_prefix: str,
    *,
    sync_listing_path: bool = True,
) -> Dict[str, Any]:
    """Update root + descendant path_slugs when an ancestor path prefix changes."""
    planned = _plan_path_prefix_cascade(session, root, old_prefix, new_prefix)
    if not planned:
        return {"updated_nodes": [], "cascade_count": 0}

    _assert_planned_path_slugs_available(session, planned)
    planned.sort(key=lambda row: row["old_path_slug"].count("/"))

    updated: List[Dict[str, Any]] = []
    for row in planned:
        node = session.get(MasterTaxonomyNode, int(row["id"]))
        if node is None:
            continue
        old_slug = row["old_path_slug"]
        new_slug = row["new_path_slug"]
        if old_slug == new_slug:
            continue
        _migrate_taxonomy_path_slug(session, node, old_slug, new_slug)
        session.execute(
            update(MasterTaxonomyNode)
            .where(MasterTaxonomyNode.id == int(node.id))
            .values(path_slug=new_slug)
        )
        session.flush()
        node = session.get(MasterTaxonomyNode, int(row["id"]))
        if node is None:
            continue
        _sync_taxonomy_node_local_paths(session, node, sync_listing_path=sync_listing_path)
        updated.append({"id": node.id, "old_path_slug": old_slug, "new_path_slug": new_slug})

    cascade_count = max(0, len(updated) - 1)
    start_shopping_prune: Dict[str, Any] = {}
    if updated:
        from db.start_shopping_config_prune import (
            path_rewrites_from_updated_nodes,
            prune_start_shopping_configs,
        )

        start_shopping_prune = prune_start_shopping_configs(
            session,
            path_rewrites=path_rewrites_from_updated_nodes(updated),
            dry_run=False,
            refresh_intersections=True,
        )
    return {
        "updated_nodes": updated,
        "cascade_count": cascade_count,
        "start_shopping_prune": start_shopping_prune,
    }


def build_node_channel_sync_params(
    node_id: int,
    payload: Dict[str, Any],
    *,
    node_ids: Optional[Sequence[int]] = None,
    include_magento: bool = True,
    include_shopify: bool = True,
) -> Dict[str, Any]:
    """Build push_nodes job params for one or more taxonomy nodes."""
    ids = [int(i) for i in (node_ids or [node_id]) if i is not None]
    if not ids:
        ids = [int(node_id)]
    return {
        "node_ids": list(dict.fromkeys(ids)),
        "magento_connection_id": payload.get("magento_connection_id"),
        "shopify_connection_id": payload.get("shopify_connection_id"),
        "dry_run": False,
        "create_missing": bool(payload.get("create_missing", True)),
        "include_magento": include_magento,
        "include_shopify": include_shopify,
        "shopify_handle_overrides": payload.get("shopify_handle_overrides"),
        "force_create_magento": bool(payload.get("force_create_magento", False)),
        "include_descendants": bool(payload.get("include_descendants", True)),
    }


def coerce_taxonomy_node_ids(node_ids: Any) -> Optional[List[int]]:
    """Normalize API/job node_ids: accept ``1``, ``\"1\"``, or ``[1, 2]``."""
    if node_ids is None:
        return None
    if isinstance(node_ids, (str, bytes)):
        text = str(node_ids).strip()
        if not text:
            return None
        return [int(text)]
    if isinstance(node_ids, (int, float)):
        return [int(node_ids)]
    if isinstance(node_ids, Sequence):
        out: List[int] = []
        for item in node_ids:
            if item is None or item == "":
                continue
            out.append(int(item))
        return out or None
    return [int(node_ids)]


def expand_taxonomy_node_ids(
    session: Session,
    node_ids: Optional[Sequence[int]],
    *,
    include_descendants: bool,
) -> Optional[List[int]]:
    """Expand requested node ids to include active descendants (parents first)."""
    normalized = coerce_taxonomy_node_ids(node_ids)
    if not normalized:
        return None
    ordered: List[int] = []
    seen: set[int] = set()
    for node_id in normalized:
        if node_id in seen:
            continue
        root = session.get(MasterTaxonomyNode, node_id)
        if root is None or not root.is_active:
            continue
        ordered.append(node_id)
        seen.add(node_id)
        if not include_descendants:
            continue
        for child in _list_descendant_nodes(session, node_id):
            if child.id in seen:
                continue
            ordered.append(int(child.id))
            seen.add(int(child.id))
    return ordered


def taxonomy_sync_node_ids(node_id: int, result: Dict[str, Any]) -> List[int]:
    """Node ids to push after upsert/reparent, including cascaded descendants."""
    cascade = result.get("path_cascade") or {}
    ids = [int(row["id"]) for row in cascade.get("updated_nodes") or [] if row.get("id") is not None]
    if int(node_id) not in ids:
        ids.insert(0, int(node_id))
    return list(dict.fromkeys(ids))


def push_taxonomy_nodes_to_channels(
    session: Session,
    *,
    node_ids: Optional[Sequence[int]] = None,
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    magento_api: Optional[Any] = None,
    shopify_client: Optional[Any] = None,
    shopify_shop_code: Optional[str] = None,
    dry_run: bool = True,
    create_missing: bool = True,
    include_magento: bool = True,
    include_shopify: bool = True,
    shopify_handle_overrides: Optional[Dict[int, str]] = None,
    force_create_magento: bool = False,
    include_descendants: bool = True,
) -> Dict[str, Any]:
    """Create or update remote Magento categories / Shopify collections for master nodes."""
    expanded_ids = expand_taxonomy_node_ids(
        session,
        node_ids,
        include_descendants=include_descendants,
    )
    stmt = select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(True)).order_by(
        MasterTaxonomyNode.sort_order, MasterTaxonomyNode.id
    )
    if expanded_ids is not None:
        stmt = stmt.where(MasterTaxonomyNode.id.in_(expanded_ids))
    nodes = list(session.scalars(stmt).all())
    if expanded_ids is not None:
        by_id = {int(node.id): node for node in nodes}
        nodes = [by_id[node_id] for node_id in expanded_ids if node_id in by_id]
    handle_overrides = {
        int(node_id): str(handle).strip()
        for node_id, handle in (shopify_handle_overrides or {}).items()
        if str(handle or "").strip()
    }

    stats = {
        "dry_run": dry_run,
        "node_count": len(nodes),
        "include_descendants": include_descendants,
        "magento_updated": 0,
        "magento_created": 0,
        "shopify_updated": 0,
        "shopify_created": 0,
        "errors": [],
    }

    for node in nodes:
        if include_magento and magento_connection_id is not None:
            from db.collection_path_guard import is_polluted_hub_collection_path

            if node.node_kind == NODE_COLLECTION and is_polluted_hub_collection_path(session, node.path_slug):
                continue
            try:
                outcome = _push_node_magento(
                    session,
                    node,
                    connection_id=magento_connection_id,
                    api=magento_api,
                    dry_run=dry_run,
                    create_missing=create_missing,
                    force_create=force_create_magento,
                )
                stats["magento_updated"] += int(outcome.get("updated") or 0)
                stats["magento_created"] += int(outcome.get("created") or 0)
            except TaxonomyChannelConflict as exc:
                stats["errors"].append(
                    {
                        "node_id": node.id,
                        "channel": exc.channel,
                        "error": str(exc),
                        "conflict": {
                            "reason": exc.reason,
                            "handle": exc.handle,
                            "path_slug": exc.path_slug,
                            "candidates": exc.candidates,
                            "related_candidates": exc.related_candidates,
                            "suggested_handles": exc.suggested_handles,
                            "guidance": exc.guidance,
                        },
                    }
                )
            except Exception as exc:
                stats["errors"].append({"node_id": node.id, "channel": "magento", "error": str(exc)})

        if include_shopify and shopify_connection_id is not None:
            from db.collection_path_guard import is_polluted_hub_collection_path

            if node.node_kind == NODE_COLLECTION and is_polluted_hub_collection_path(session, node.path_slug):
                continue
            try:
                outcome = _push_node_shopify(
                    session,
                    node,
                    connection_id=shopify_connection_id,
                    client=shopify_client,
                    shop_code=shopify_shop_code,
                    dry_run=dry_run,
                    create_missing=create_missing,
                    handle_override=handle_overrides.get(node.id),
                )
                stats["shopify_updated"] += int(outcome.get("updated") or 0)
                stats["shopify_created"] += int(outcome.get("created") or 0)
                stats["shopify_linked"] = int(stats.get("shopify_linked") or 0) + int(outcome.get("linked") or 0)
            except TaxonomyChannelConflict as exc:
                stats["errors"].append(
                    {
                        "node_id": node.id,
                        "channel": exc.channel,
                        "error": str(exc),
                        "conflict": {
                            "reason": exc.reason,
                            "handle": exc.handle,
                            "path_slug": exc.path_slug,
                            "candidates": exc.candidates,
                            "related_candidates": exc.related_candidates,
                            "suggested_handles": exc.suggested_handles,
                            "guidance": exc.guidance,
                        },
                    }
                )
            except Exception as exc:
                stats["errors"].append({"node_id": node.id, "channel": "shopify", "error": str(exc)})

    if not dry_run:
        session.flush()
    stats["error_count"] = len(stats["errors"])
    return stats


def assign_skus_to_taxonomy_node(
    session: Session,
    *,
    node_id: int,
    skus: Optional[Sequence[str]] = None,
    only_assigned: bool = False,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    search: Optional[str] = None,
    replace_existing: bool = False,
    dry_run: bool = True,
    limit: Optional[int] = None,
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
) -> Dict[str, Any]:
    """Assign master SKUs to a taxonomy node locally (listing paths + channel taxonomy rows)."""
    from db.master_taxonomy_merge import resolve_canonical_node

    node = resolve_canonical_node(session, node_id=node_id) or session.get(MasterTaxonomyNode, int(node_id))
    if node is None or not node.is_active:
        raise ValueError(f"Unknown taxonomy node: {node_id}")

    from db.channel_listing_path import bulk_assign_listing_paths
    from db.product_channel_taxonomy import resolve_filtered_master_skus

    master_skus = list(
        skus
        or resolve_filtered_master_skus(
            session,
            skus=None,
            only_assigned=only_assigned,
            category_l1=category_l1,
            category_l2=category_l2,
            category_l3=category_l3,
            collection=collection,
            product_family=product_family,
            search=search,
            limit=limit,
        )
    )
    if not master_skus:
        return {"node_id": node_id, "sku_count": 0, "assignment_count": 0}

    listing = bulk_assign_listing_paths(
        session,
        path_slugs=[node.path_slug],
        skus=master_skus,
        dry_run=dry_run,
        replace_existing=replace_existing,
    )
    assignment_count = int(
        listing.get("would_upsert")
        or listing.get("upserted")
        or listing.get("assignment_count")
        or 0
    )

    links = _links_for_node(session, node.id)
    channel_results: Dict[str, Any] = {}
    magento_link = links.get("magento")
    magento_kind = "collection" if node.node_kind == NODE_COLLECTION else MAGENTO_KIND
    if magento_connection_id is not None:
        preferred_magento = _channel_link(
            session,
            int(node.id),
            "magento",
            int(magento_connection_id),
            magento_kind,
        )
        if preferred_magento is None and magento_kind != MAGENTO_KIND:
            preferred_magento = _channel_link(
                session,
                int(node.id),
                "magento",
                int(magento_connection_id),
                MAGENTO_KIND,
            )
        if preferred_magento is not None:
            magento_link = preferred_magento
    if magento_link and magento_link.remote_id:
        magento_conn = (
            int(magento_connection_id)
            if magento_connection_id is not None
            else magento_link.connection_id
        )
        channel_results["magento"] = bulk_assign_sku_taxonomy(
            session,
            channel_code="magento",
            remote_ids=[magento_link.remote_id],
            taxonomy_kind=magento_link.taxonomy_kind or magento_kind,
            connection_id=magento_conn,
            skus=master_skus,
            remote_path=magento_link.remote_path or node.name,
            replace_existing=replace_existing,
            dry_run=dry_run,
        )
        assignment_count += int(channel_results["magento"].get("assignment_count") or 0)

    shopify_link = links.get("shopify")
    if shopify_connection_id is not None:
        preferred_shopify = _channel_link(
            session,
            int(node.id),
            "shopify",
            int(shopify_connection_id),
            SHOPIFY_COLLECTION_KIND,
        )
        if preferred_shopify is not None:
            shopify_link = preferred_shopify
    if shopify_link and shopify_link.remote_id:
        shopify_conn = (
            int(shopify_connection_id)
            if shopify_connection_id is not None
            else shopify_link.connection_id
        )
        channel_results["shopify"] = bulk_assign_sku_taxonomy(
            session,
            channel_code="shopify",
            remote_ids=[shopify_link.remote_id],
            taxonomy_kind=SHOPIFY_COLLECTION_KIND,
            connection_id=shopify_conn,
            skus=master_skus,
            remote_path=shopify_link.remote_path or node.name,
            replace_existing=replace_existing,
            dry_run=dry_run,
        )
        assignment_count += int(channel_results["shopify"].get("assignment_count") or 0)

    if not dry_run and node.node_kind == NODE_COLLECTION:
        leaf = _leaf_title(node.name)
        for sku in master_skus:
            product = session.scalar(select(MasterProduct).where(MasterProduct.sku == sku).limit(1))
            if product:
                product.collection = leaf
        session.flush()

    return {
        "node_id": node_id,
        "path_slug": node.path_slug,
        "sku_count": len(master_skus),
        "assignment_count": assignment_count,
        "listing_paths": listing,
        "channels": channel_results,
        "dry_run": dry_run,
        "skus": master_skus,
        "skus_sample": master_skus[:50],
        "magento_connection_id": magento_connection_id,
        "shopify_connection_id": shopify_connection_id,
    }


def _listing_path_target_remote_id(
    session: Session,
    path_slug: str,
    *,
    connection_id: int,
    taxonomy_kind: str,
) -> Optional[str]:
    slug = normalize_plp_path(path_slug)
    path = session.scalar(select(ChannelListingPath).where(ChannelListingPath.path_slug == slug).limit(1))
    if path is None:
        return None
    targets = session.scalars(
        select(ChannelListingPathTarget)
        .where(ChannelListingPathTarget.listing_path_id == path.id)
        .where(ChannelListingPathTarget.channel_code == "magento")
        .where(ChannelListingPathTarget.taxonomy_kind == taxonomy_kind)
        .where(ChannelListingPathTarget.is_active.is_(True))
        .where(
            ChannelListingPathTarget.connection_id.is_(None)
            | (ChannelListingPathTarget.connection_id == int(connection_id))
        )
        .order_by(ChannelListingPathTarget.connection_id.desc().nulls_last())
    ).all()
    for target in targets:
        remote_id = str(target.remote_id or "").strip()
        if remote_id.isdigit():
            return remote_id
    return None


def _magento_remote_id_by_url_key(api: Any, path_slug: str) -> Optional[int]:
    leaf = normalize_plp_path(path_slug).split("/")[-1].strip()
    if not leaf or api is None:
        return None
    status, items = api.search_categories(url_key=leaf)
    if status != 200 or not items:
        return None
    matches = [item for item in items if str(item.get("id") or "").isdigit()]
    if len(matches) != 1:
        return None
    return int(matches[0]["id"])


def _resolve_magento_taxonomy_remote_id(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    connection_id: int,
    taxonomy_kind: str,
    api: Optional[Any],
    link: Optional[MasterTaxonomyChannelLink],
    guidance: Optional[Dict[str, Any]] = None,
) -> Optional[int]:
    if link and str(link.remote_id or "").strip().isdigit():
        return int(link.remote_id)

    listing_remote = _listing_path_target_remote_id(
        session,
        node.path_slug,
        connection_id=connection_id,
        taxonomy_kind=taxonomy_kind,
    )
    if listing_remote:
        return int(listing_remote)

    if api is not None:
        url_key_match = _magento_remote_id_by_url_key(api, node.path_slug)
        if url_key_match is not None:
            return url_key_match

    canonical_rows = [
        row
        for row in (guidance or {}).get("canonical_candidates") or []
        if row.get("is_canonical_path") and str(row.get("category_id") or row.get("remote_id") or "").isdigit()
    ]
    if len(canonical_rows) == 1:
        return int(canonical_rows[0].get("category_id") or canonical_rows[0]["remote_id"])

    return None


def _is_magento_category_missing(api: Any, category_id: int) -> bool:
    """Return True when Magento confirms the category id does not exist."""
    getter = getattr(api, "get_category", None)
    if callable(getter):
        status, _body = getter(int(category_id))
        if status == 404:
            return True
        if status == 200:
            return False
    return False


def _deactivate_stale_magento_remote_for_connection(
    session: Session,
    *,
    connection_id: int,
    remote_id: str,
    link: Optional[MasterTaxonomyChannelLink] = None,
) -> None:
    """Clear local mappings that point at a Magento category that no longer exists."""
    remote_id = str(remote_id or "").strip()
    if not remote_id:
        return

    if link is not None and link.is_active:
        link.is_active = False

    for row in session.scalars(
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.channel_code == "magento")
        .where(MasterTaxonomyChannelLink.connection_id == int(connection_id))
        .where(MasterTaxonomyChannelLink.remote_id == remote_id)
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
    ).all():
        row.is_active = False

    for target in session.scalars(
        select(ChannelListingPathTarget)
        .where(ChannelListingPathTarget.channel_code == "magento")
        .where(ChannelListingPathTarget.remote_id == remote_id)
        .where(ChannelListingPathTarget.is_active.is_(True))
        .where(
            ChannelListingPathTarget.connection_id.is_(None)
            | (ChannelListingPathTarget.connection_id == int(connection_id))
        )
    ).all():
        target.is_active = False

    if remote_id.isdigit():
        for reg in session.scalars(
            select(MagentoCategoryRegistry)
            .where(MagentoCategoryRegistry.connection_id == int(connection_id))
            .where(MagentoCategoryRegistry.category_id == int(remote_id))
            .where(MagentoCategoryRegistry.is_active.is_(True))
        ).all():
            reg.is_active = False

    session.flush()


def _push_node_magento(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    connection_id: int,
    api: Optional[Any],
    dry_run: bool,
    create_missing: bool,
    force_create: bool = False,
) -> Dict[str, int]:
    title = _leaf_title(node.name)
    parent_remote = _magento_parent_id(session, node, connection_id)
    kind = MAGENTO_KIND if node.node_kind != NODE_COLLECTION else "collection"
    link = _channel_link(session, node.id, "magento", connection_id, kind)
    guidance = magento_taxonomy_channel_guidance(session, node, connection_id)

    if dry_run:
        remote_id = _resolve_magento_taxonomy_remote_id(
            session,
            node,
            connection_id=connection_id,
            taxonomy_kind=kind,
            api=None,
            link=link,
            guidance=guidance,
        )
        return {"updated": int(bool(remote_id)), "created": int(not remote_id and create_missing)}

    remote_id = _resolve_magento_taxonomy_remote_id(
        session,
        node,
        connection_id=connection_id,
        taxonomy_kind=kind,
        api=api,
        link=link,
        guidance=guidance,
    )

    if remote_id is not None and api is not None:
        stale = _is_magento_category_missing(api, int(remote_id))
        if not stale:
            status, body = api.update_category(
                int(remote_id),
                {"id": int(remote_id), "name": title, "parent_id": parent_remote, "is_active": True},
            )
            if status == 404:
                stale = True
            elif status not in (200, 201):
                raise RuntimeError(f"Magento update_category failed: HTTP {status} {body}")
            else:
                _upsert_magento_registry(session, connection_id, int(remote_id), parent_remote, title, node)
                if link is None:
                    link = _upsert_channel_link(
                        session,
                        node_id=node.id,
                        channel_code="magento",
                        connection_id=connection_id,
                        taxonomy_kind=kind,
                        remote_id=str(remote_id),
                        remote_parent_id=str(parent_remote),
                        remote_path=title,
                    )
                else:
                    link.remote_parent_id = str(parent_remote)
                    link.remote_path = title
                _write_listing_target(session, node, "magento", connection_id, str(remote_id), kind, title)
                _push_taxonomy_category_icon_magento(session, api, node, int(remote_id), dry_run=False)
                return {"updated": 1, "created": 0}

        if stale:
            _deactivate_stale_magento_remote_for_connection(
                session,
                connection_id=connection_id,
                remote_id=str(remote_id),
                link=link,
            )
            link = None
            remote_id = None
            # Dead Magento ids must recreate under the parent, not block on stale registry matches.
            force_create = True
            guidance = magento_taxonomy_channel_guidance(session, node, connection_id)

    if not create_missing or api is None:
        return {"updated": 0, "created": 0}

    from db.taxonomy_invent_policy import can_provision_remotes

    if not can_provision_remotes():
        return {
            "updated": 0,
            "created": 0,
            "skipped_frozen": 1,
            "message": "Remote taxonomy provision is frozen; Magento create skipped",
        }

    if not force_create:
        action = guidance.get("recommended_action")
        if action == "link_parent_hub":
            parent_name = guidance.get("parent_hub_name") or "parent hub"
            _raise_magento_taxonomy_conflict(
                session,
                node,
                connection_id=connection_id,
                reason="parent_hub_unlinked",
                message=f"Link Magento category on parent node {parent_name!r} before creating {title!r}.",
                guidance=guidance,
            )
        if action == "link_canonical":
            _raise_magento_taxonomy_conflict(
                session,
                node,
                connection_id=connection_id,
                reason="canonical_exists",
                message=f"Magento already has a category at the expected path for {title!r}. Link it below.",
                guidance=guidance,
            )
        if action == "create_under_parent" and guidance.get("wrong_branch_candidates"):
            pass
        elif guidance.get("wrong_branch_candidates"):
            _raise_magento_taxonomy_conflict(
                session,
                node,
                connection_id=connection_id,
                reason="wrong_branch_only",
                message=(
                    f"Found {len(guidance['wrong_branch_candidates'])} Magento "
                    f"categories named {title!r} under the wrong parent. "
                    f"Create a new one under {guidance.get('parent_hub_name') or 'the parent hub'}."
                ),
                guidance=guidance,
            )
    elif node.parent_id and not guidance.get("parent_hub_magento_linked"):
        parent_name = guidance.get("parent_hub_name") or "parent hub"
        raise RuntimeError(f"Link Magento category on parent node {parent_name!r} before force-creating {title!r}.")

    status, body = api.create_category(
        {"parent_id": parent_remote, "name": title, "is_active": True, "include_in_menu": True}
    )
    if status not in (200, 201) or not isinstance(body, dict) or not body.get("id"):
        if status == 400 and guidance.get("wrong_branch_candidates"):
            _raise_magento_taxonomy_conflict(
                session,
                node,
                connection_id=connection_id,
                reason="create_blocked",
                message=f"Magento blocked create for {title!r}. Use create under parent or link after pull.",
                guidance=guidance,
            )
        raise RuntimeError(f"Magento create_category failed: HTTP {status} {body}")
    category_id = int(body["id"])
    _upsert_magento_registry(session, connection_id, category_id, parent_remote, title, node)
    _upsert_channel_link(
        session,
        node_id=node.id,
        channel_code="magento",
        connection_id=connection_id,
        taxonomy_kind=kind,
        remote_id=str(category_id),
        remote_parent_id=str(parent_remote),
        remote_path=title,
    )
    _write_listing_target(session, node, "magento", connection_id, str(category_id), kind, title)
    _push_taxonomy_category_icon_magento(session, api, node, category_id, dry_run=False)
    return {"updated": 0, "created": 1}


def _ensure_listing_path_for_node(session: Session, node: MasterTaxonomyNode) -> None:
    """Ensure channel_listing_path exists before writing channel targets."""
    upsert_listing_path(
        session,
        path_slug=node.path_slug,
        title=node.name,
        path_kind=node.node_kind,
        parent_path_slug=_parent_path_slug(session, node.parent_id),
        notes="master taxonomy channel link",
    )


def _title_from_path_slug(path_slug: str) -> str:
    slug = normalize_plp_path(path_slug or "")
    if not slug:
        return ""
    return " ".join(_title_from_slug_segment(seg) for seg in slug.split("/") if seg)


def _title_from_slug_segment(segment: str) -> str:
    words = str(segment or "").replace("-", " ").split()
    return " ".join(word if word.lower() == "and" else word.capitalize() for word in words)


def _taxonomy_ancestor_chain(session: Session, node: MasterTaxonomyNode) -> List[MasterTaxonomyNode]:
    chain: List[MasterTaxonomyNode] = []
    current: Optional[MasterTaxonomyNode] = node
    seen: set[int] = set()
    while current and current.id not in seen:
        seen.add(current.id)
        chain.append(current)
        if not current.parent_id:
            break
        current = session.get(MasterTaxonomyNode, int(current.parent_id))
    chain.reverse()
    return chain


def _shopify_title_for_node(session: Session, node: MasterTaxonomyNode) -> str:
    """Shopify collections are flat — use the full taxonomy path as the display title."""
    chain = _taxonomy_ancestor_chain(session, node)
    if len(chain) > 1:
        return " ".join(_leaf_title(row.name) for row in chain)
    slug = normalize_plp_path(node.path_slug or "")
    if slug and "/" in slug:
        return _title_from_path_slug(slug)
    if chain:
        return _leaf_title(chain[0].name)
    return _title_from_path_slug(slug) or _leaf_title(node.name)


def _normalize_shopify_handle(value: str) -> str:
    import re

    text = str(value or "").strip().lower()
    text = re.sub(r"[^a-z0-9_-]+", "-", text).strip("-")
    return text[:250] or "collection"


def _shopify_leaf_handle(node: MasterTaxonomyNode) -> str:
    from db.catalog_intent_planner import slugify_segment

    slug = normalize_plp_path(node.path_slug or "")
    if not slug:
        return slugify_segment(_leaf_title(node.name))
    return slug.rsplit("/", 1)[-1]


def _shopify_handle_for_node(node: MasterTaxonomyNode, *, override: Optional[str] = None) -> str:
    from db.catalog_intent_planner import slugify_segment

    if override:
        return _normalize_shopify_handle(override)
    slug = normalize_plp_path(node.path_slug or "")
    if slug:
        return slug.replace("/", "-")
    return slugify_segment(_leaf_title(node.name))


def _shopify_handle_suggestions(node: MasterTaxonomyNode) -> List[str]:
    from db.catalog_intent_planner import slugify_segment

    slug = normalize_plp_path(node.path_slug or "")
    if not slug:
        return [slugify_segment(_leaf_title(node.name))]
    seen: set[str] = set()
    out: List[str] = []

    def add(value: str) -> None:
        clean = _normalize_shopify_handle(value)
        if clean and clean not in seen:
            seen.add(clean)
            out.append(clean)

    add(slug.replace("/", "-"))
    parts = [part for part in slug.split("/") if part]
    if len(parts) > 1:
        add(f"{parts[-2]}-{parts[-1]}")
        root_prefix = parts[0].split("-")[0]
        add(f"{root_prefix}-{parts[-1]}")
        underscore = f"{root_prefix}_{parts[-1]}"
        if underscore not in seen:
            seen.add(underscore)
            out.append(underscore)
    return out


def _shopify_conflict_candidate_sets(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    connection_id: int,
    preferred_handle: str,
    title: str,
    client: Optional[Any],
) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    from shopify.collection_sync import find_shopify_collection_candidates

    leaf_handle = _shopify_leaf_handle(node)
    blocking_raw = find_shopify_collection_candidates(
        session,
        connection_id,
        handle=preferred_handle,
        title=title,
        client=None,
    )
    blocking = [
        {
            **_enrich_channel_candidate(session, "shopify", connection_id, row, exclude_node_id=node.id),
            "match_kind": "exact_handle",
        }
        for row in blocking_raw
    ]
    related: List[Dict[str, Any]] = []
    if leaf_handle and leaf_handle != preferred_handle:
        related_raw = find_shopify_collection_candidates(
            session,
            connection_id,
            handle=leaf_handle,
            title=title,
            client=client,
        )
        seen_ids = {row.get("remote_id") for row in blocking}
        for row in related_raw:
            remote_id = row.get("remote_id")
            if remote_id in seen_ids:
                continue
            enriched = _enrich_channel_candidate(
                session, "shopify", connection_id, row, exclude_node_id=node.id
            )
            enriched["match_kind"] = "same_leaf_different_branch"
            related.append(enriched)
    return blocking, related


def _raise_shopify_handle_conflict(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    connection_id: int,
    preferred_handle: str,
    title: str,
    client: Optional[Any],
    reason: str,
    message: str,
) -> None:
    blocking, related = _shopify_conflict_candidate_sets(
        session,
        node,
        connection_id=connection_id,
        preferred_handle=preferred_handle,
        title=title,
        client=client,
    )
    suggestions = _shopify_handle_suggestions(node)
    raise TaxonomyChannelConflict(
        channel="shopify",
        reason=reason,
        handle=preferred_handle,
        message=message,
        candidates=blocking,
        related_candidates=related,
        suggested_handles=suggestions,
        path_slug=node.path_slug,
    )


def _push_node_shopify(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    connection_id: int,
    client: Optional[Any],
    shop_code: Optional[str],
    dry_run: bool,
    create_missing: bool,
    handle_override: Optional[str] = None,
) -> Dict[str, int]:
    title = _shopify_title_for_node(session, node)
    handle = _shopify_handle_for_node(node, override=handle_override)
    link = _channel_link(session, node.id, "shopify", connection_id, SHOPIFY_COLLECTION_KIND)

    if dry_run:
        return {"updated": int(bool(link)), "created": int(not link and create_missing)}

    if link and link.remote_id and client is not None:
        from shopify.collection_landing_push import COLLECTION_UPDATE

        data = client.graphql(
            COLLECTION_UPDATE,
            {"input": {"id": link.remote_id, "title": title, "handle": handle}},
        )
        result = data.get("collectionUpdate") or {}
        errors = result.get("userErrors") or []
        if errors:
            raise RuntimeError(f"collectionUpdate userErrors: {errors}")
        collection = result.get("collection") or {}
        _upsert_shopify_registry(session, connection_id, shop_code, link.remote_id, collection, title)
        link.remote_path = title
        _write_listing_target(
            session, node, "shopify", connection_id, link.remote_id, SHOPIFY_COLLECTION_KIND, title
        )
        return {"updated": 1, "created": 0}

    if not create_missing or client is None or not shop_code:
        return {"updated": 0, "created": 0}

    from db.taxonomy_invent_policy import can_provision_remotes

    if not can_provision_remotes():
        return {
            "updated": 0,
            "created": 0,
            "skipped_frozen": 1,
            "message": "Remote taxonomy provision is frozen; Shopify create skipped",
        }

    from shopify.collection_sync import create_shopify_collection

    blocking, related = _shopify_conflict_candidate_sets(
        session,
        node,
        connection_id=connection_id,
        preferred_handle=handle,
        title=title,
        client=client,
    )
    if blocking:
        _raise_shopify_handle_conflict(
            session,
            node,
            connection_id=connection_id,
            preferred_handle=handle,
            title=title,
            client=client,
            reason="already_exists",
            message=f"Shopify collection handle {handle!r} already exists",
        )

    try:
        created = create_shopify_collection(
            session,
            client,
            shop_code=shop_code,
            connection_id=connection_id,
            title=title,
            handle=handle,
        )
    except RuntimeError as exc:
        if _is_shopify_handle_taken_error(exc):
            _raise_shopify_handle_conflict(
                session,
                node,
                connection_id=connection_id,
                preferred_handle=handle,
                title=title,
                client=client,
                reason="handle_taken",
                message=str(exc),
            )
        raise

    collection_id = created["collection_id"]
    _upsert_channel_link(
        session,
        node_id=node.id,
        channel_code="shopify",
        connection_id=connection_id,
        taxonomy_kind=SHOPIFY_COLLECTION_KIND,
        remote_id=collection_id,
        remote_parent_id="0",
        remote_path=title,
    )
    _write_listing_target(session, node, "shopify", connection_id, collection_id, SHOPIFY_COLLECTION_KIND, title)
    return {"updated": 0, "created": 1}


def magento_taxonomy_channel_guidance(
    session: Session,
    node: MasterTaxonomyNode,
    connection_id: int,
) -> Dict[str, Any]:
    """How to resolve Magento placement for a taxonomy node (link vs create under parent)."""
    from db.master_taxonomy_hierarchy import _magento_path_candidates, magento_registry_candidates_for_node

    registry_rows = _magento_registry_rows(session, connection_id)
    scored = magento_registry_candidates_for_node(node, registry_rows)
    canonical_rows = [row for row in scored if row.get("is_canonical_path")]
    wrong_branch_rows = [row for row in scored if not row.get("is_canonical_path")]

    parent_node = session.get(MasterTaxonomyNode, int(node.parent_id)) if node.parent_id else None
    parent_link = None
    if parent_node is not None:
        parent_kind = MAGENTO_KIND if parent_node.node_kind != NODE_COLLECTION else "collection"
        parent_link = _channel_link(session, parent_node.id, "magento", connection_id, parent_kind)

    path_candidates = _magento_path_candidates(node.path_slug, node.name)
    expected_path = path_candidates[0] if path_candidates else _leaf_title(node.name)

    if canonical_rows:
        recommended_action = "link_canonical"
    elif parent_node is not None and not (parent_link and parent_link.remote_id):
        recommended_action = "link_parent_hub"
    elif wrong_branch_rows:
        recommended_action = "create_under_parent"
    else:
        recommended_action = "create_under_parent"

    from db.magento_path_names import title_path_to_magento_path_names

    parent_hub_name = None
    if parent_node is not None:
        # Prefer Magento-facing Tall Cabinets over legacy master short name "Tall".
        parent_hub_name = title_path_to_magento_path_names(parent_node.name).replace("/", " / ")

    return {
        "recommended_action": recommended_action,
        "has_canonical_candidate": bool(canonical_rows),
        "expected_magento_path": expected_path,
        "parent_hub_node_id": parent_node.id if parent_node else None,
        "parent_hub_name": parent_hub_name,
        "parent_hub_magento_linked": bool(parent_link and parent_link.remote_id),
        "parent_hub_magento_remote_id": str(parent_link.remote_id) if parent_link and parent_link.remote_id else None,
        "canonical_candidates": canonical_rows,
        "wrong_branch_candidates": wrong_branch_rows,
    }


def _raise_magento_taxonomy_conflict(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    connection_id: int,
    reason: str,
    message: str,
    guidance: Dict[str, Any],
) -> None:
    title = _leaf_title(node.name)
    canonical = [
        {**_enrich_channel_candidate(session, "magento", connection_id, item, exclude_node_id=node.id), "linkable": True}
        for item in guidance.get("canonical_candidates") or []
    ]
    wrong_branch = [
        {
            **_enrich_channel_candidate(session, "magento", connection_id, item, exclude_node_id=node.id),
            "linkable": False,
            "is_canonical_path": False,
        }
        for item in guidance.get("wrong_branch_candidates") or []
    ]
    raise TaxonomyChannelConflict(
        channel="magento",
        reason=reason,
        handle=title,
        message=message,
        candidates=canonical if canonical else wrong_branch,
        related_candidates=wrong_branch if canonical else [],
        path_slug=node.path_slug,
        guidance=guidance,
    )


def find_taxonomy_channel_candidates(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    channel_code: str,
    connection_id: int,
    client: Optional[Any] = None,
) -> Dict[str, Any]:
    """Return existing remote channel entities that may match this taxonomy node."""
    channel_code = str(channel_code or "").strip().lower()
    title = _shopify_title_for_node(session, node)
    handle = _shopify_handle_for_node(node)
    leaf_handle = _shopify_leaf_handle(node)
    candidates: List[Dict[str, Any]] = []
    related_candidates: List[Dict[str, Any]] = []
    guidance: Dict[str, Any] = {}

    if channel_code == "shopify":
        blocking, related = _shopify_conflict_candidate_sets(
            session,
            node,
            connection_id=connection_id,
            preferred_handle=handle,
            title=title,
            client=client,
        )
        candidates = blocking
        related_candidates = related
    elif channel_code == "magento":
        guidance = magento_taxonomy_channel_guidance(session, node, connection_id)
        canonical = guidance.get("canonical_candidates") or []
        wrong_branch = guidance.get("wrong_branch_candidates") or []
        candidates = [
            _enrich_channel_candidate(
                session,
                channel_code,
                connection_id,
                {
                    "remote_id": item["remote_id"],
                    "title": item.get("name") or item.get("path_names") or title,
                    "path_names": item.get("path_names"),
                    "parent_id": item.get("parent_id"),
                    "source": "registry",
                    "is_canonical_path": True,
                    "match_reason": "path",
                    "duplicate_group_size": item.get("duplicate_group_size"),
                    "linkable": True,
                },
                exclude_node_id=node.id,
            )
            for item in canonical
        ]
        related_candidates = [
            _enrich_channel_candidate(
                session,
                channel_code,
                connection_id,
                {
                    "remote_id": item["remote_id"],
                    "title": item.get("name") or item.get("path_names") or title,
                    "path_names": item.get("path_names"),
                    "parent_id": item.get("parent_id"),
                    "source": "registry",
                    "is_canonical_path": False,
                    "match_reason": item.get("match_reason"),
                    "duplicate_group_size": item.get("duplicate_group_size"),
                    "linkable": bool(canonical),
                },
                exclude_node_id=node.id,
            )
            for item in wrong_branch
        ]
    else:
        raise ValueError(f"Unsupported channel_code: {channel_code}")

    return {
        "node_id": node.id,
        "channel_code": channel_code,
        "handle": handle,
        "leaf_handle": leaf_handle,
        "path_slug": node.path_slug,
        "title": title,
        "candidates": candidates,
        "related_candidates": related_candidates,
        "suggested_handles": _shopify_handle_suggestions(node) if channel_code == "shopify" else [],
        "candidate_count": len(candidates),
        "guidance": guidance,
    }


def reconcile_magento_taxonomy_duplicates(
    session: Session,
    *,
    connection_id: int,
    node_kind: Optional[str] = None,
    limit: Optional[int] = 500,
) -> Dict[str, Any]:
    """Surface master taxonomy nodes whose Magento leaf name exists in multiple registry paths."""
    from db.master_taxonomy_hierarchy import (
        NODE_COLLECTION,
        magento_registry_candidates_for_node,
    )

    registry_rows = _magento_registry_rows(session, connection_id)
    stmt = (
        select(MasterTaxonomyNode)
        .where(MasterTaxonomyNode.is_active.is_(True))
        .order_by(MasterTaxonomyNode.path_slug)
    )
    kind = str(node_kind or NODE_COLLECTION).strip() or NODE_COLLECTION
    stmt = stmt.where(MasterTaxonomyNode.node_kind == kind)
    if limit:
        stmt = stmt.limit(int(limit))

    items: List[Dict[str, Any]] = []
    duplicate_nodes = 0
    unlinked_nodes = 0
    missing_in_registry = 0
    resolved_nodes = 0
    needs_review_nodes = 0
    for node in session.scalars(stmt).all():
        scored = magento_registry_candidates_for_node(node, registry_rows)
        link = session.scalar(
            select(MasterTaxonomyChannelLink)
            .where(MasterTaxonomyChannelLink.taxonomy_node_id == node.id)
            .where(MasterTaxonomyChannelLink.channel_code == "magento")
            .where(MasterTaxonomyChannelLink.connection_id == connection_id)
            .where(MasterTaxonomyChannelLink.is_active.is_(True))
            .limit(1)
        )
        linked_remote_id = str(link.remote_id).strip() if link and link.remote_id else None
        if not linked_remote_id:
            unlinked_nodes += 1
        if not scored:
            missing_in_registry += 1
            needs_review = not linked_remote_id
            if needs_review:
                needs_review_nodes += 1
            else:
                resolved_nodes += 1
            if needs_review:
                items.append(
                    {
                        "node_id": node.id,
                        "node_kind": node.node_kind,
                        "name": node.name,
                        "path_slug": node.path_slug,
                        "linked_remote_id": linked_remote_id,
                        "has_duplicates": False,
                        "duplicate_count": 0,
                        "missing_in_registry": True,
                        "needs_review": True,
                        "is_correctly_linked": False,
                        "recommended_remote_id": None,
                        "recommended_path": None,
                        "recommended_is_canonical": False,
                        "candidates": [],
                    }
                )
            continue
        has_duplicates = len(scored) > 1
        recommended = scored[0]
        recommended_remote_id = str(recommended.get("remote_id") or "").strip() or None
        recommended_is_canonical = bool(recommended.get("is_canonical_path"))
        is_correctly_linked = bool(
            linked_remote_id
            and recommended_remote_id
            and linked_remote_id == recommended_remote_id
            and recommended_is_canonical
        )
        needs_review = not is_correctly_linked and (
            not linked_remote_id
            or missing_in_registry
            or has_duplicates
            or (linked_remote_id != recommended_remote_id)
        )
        if is_correctly_linked:
            resolved_nodes += 1
        elif needs_review:
            needs_review_nodes += 1
            if has_duplicates:
                duplicate_nodes += 1
            items.append(
                {
                    "node_id": node.id,
                    "node_kind": node.node_kind,
                    "name": node.name,
                    "path_slug": node.path_slug,
                    "linked_remote_id": linked_remote_id,
                    "has_duplicates": has_duplicates,
                    "duplicate_count": len(scored),
                    "missing_in_registry": False,
                    "needs_review": True,
                    "is_correctly_linked": False,
                    "recommended_remote_id": recommended_remote_id,
                    "recommended_path": recommended.get("stripped_path") or recommended.get("path_names"),
                    "recommended_is_canonical": recommended_is_canonical,
                    "candidates": scored,
                }
            )

    return {
        "connection_id": connection_id,
        "node_kind": kind,
        "item_count": len(items),
        "duplicate_nodes": duplicate_nodes,
        "unlinked_nodes": unlinked_nodes,
        "missing_in_registry": missing_in_registry,
        "resolved_nodes": resolved_nodes,
        "needs_review_nodes": needs_review_nodes,
        "items": items,
    }


def link_taxonomy_node_channel(
    session: Session,
    node_id: int,
    *,
    channel_code: str,
    remote_id: str,
    connection_id: int,
    shop_code: Optional[str] = None,
) -> Dict[str, Any]:
    """Associate an existing remote Magento category or Shopify collection with a taxonomy node."""
    node = session.get(MasterTaxonomyNode, int(node_id))
    if node is None or not node.is_active:
        raise ValueError(f"Unknown taxonomy node: {node_id}")

    channel_code = str(channel_code or "").strip().lower()
    remote_id = str(remote_id or "").strip()
    if not remote_id:
        raise ValueError("remote_id is required")

    other = _remote_linked_taxonomy_node(
        session,
        channel_code,
        connection_id,
        remote_id,
        exclude_node_id=node.id,
    )
    if other is not None:
        raise ValueError(
            f"Remote ID already linked to taxonomy node {other.name!r} (id={other.id})"
        )

    title = _leaf_title(node.name)
    if channel_code == "shopify":
        title = _shopify_title_for_node(session, node)
        reg = session.scalar(
            select(ShopifyCollectionRegistry)
            .where(ShopifyCollectionRegistry.collection_id == remote_id)
            .limit(1)
        )
        collection_meta = {
            "id": remote_id,
            "handle": reg.handle if reg else _shopify_handle_for_node(node),
            "title": reg.title if reg else title,
        }
        _link_shopify_collection_to_node(
            session,
            node,
            connection_id=connection_id,
            shop_code=shop_code,
            collection_id=remote_id,
            collection=collection_meta,
            title=collection_meta["title"] or title,
        )
    elif channel_code == "magento":
        kind = MAGENTO_KIND if node.node_kind != NODE_COLLECTION else "collection"
        parent_remote = _magento_parent_id(session, node, connection_id)
        reg_rows = _magento_registry_rows(session, connection_id)
        match = next((r for r in reg_rows if str(r.get("category_id")) == remote_id), None)
        remote_title = (match or {}).get("name") or (match or {}).get("path_names") or title
        _upsert_magento_registry(
            session,
            connection_id,
            int(remote_id),
            parent_remote,
            remote_title,
            node,
        )
        _upsert_channel_link(
            session,
            node_id=node.id,
            channel_code="magento",
            connection_id=connection_id,
            taxonomy_kind=kind,
            remote_id=remote_id,
            remote_parent_id=str(parent_remote),
            remote_path=remote_title,
        )
        _write_listing_target(session, node, "magento", connection_id, remote_id, kind, remote_title)
    else:
        raise ValueError(f"Unsupported channel_code: {channel_code}")

    session.flush()
    links = _links_for_node(session, node.id)
    return {"linked": True, "node": _node_payload(node, links)}


def _default_path_slug(session: Session, name: str, parent_id: Optional[int]) -> str:
    from db.catalog_intent_planner import slugify_segment

    leaf = slugify_segment(_leaf_title(name))
    if not parent_id:
        return leaf
    parent = session.get(MasterTaxonomyNode, int(parent_id))
    if parent and parent.path_slug:
        return f"{parent.path_slug}/{leaf}"
    return leaf


def _ensure_collection_registry(
    session: Session,
    name: str,
    parent_id: Optional[int],
    *,
    allow_invent: bool = False,
) -> Optional[int]:
    from db.master_taxonomy_registry import upsert_collection_registry

    category_l1 = "Cabinets"
    if parent_id:
        parent = session.get(MasterTaxonomyNode, int(parent_id))
        if parent:
            category_l1 = _leaf_title(parent.name)
    leaf = _leaf_title(name)
    code = registry_code_from_name(name)
    existing = session.scalar(
        select(MasterCollectionRegistry)
        .where(MasterCollectionRegistry.is_active.is_(True))
        .where(
            (MasterCollectionRegistry.code == code) | (MasterCollectionRegistry.name == leaf)
        )
        .limit(1)
    )
    if existing is not None:
        return int(existing.id)
    if not allow_invent:
        return None
    created = upsert_collection_registry(
        session,
        {"name": leaf, "code": code, "category_l1": category_l1},
        allow_invent=True,
    )
    return int(created["id"])


def _magento_parent_id(session: Session, node: MasterTaxonomyNode, connection_id: int) -> int:
    if not node.parent_id:
        return MAGENTO_ROOT_CATEGORY_ID
    parent_link = _channel_link(session, int(node.parent_id), "magento", connection_id, MAGENTO_KIND)
    if parent_link and parent_link.remote_id:
        return int(parent_link.remote_id)
    return MAGENTO_ROOT_CATEGORY_ID


def _channel_link(
    session: Session,
    node_id: int,
    channel_code: str,
    connection_id: int,
    taxonomy_kind: str,
) -> Optional[MasterTaxonomyChannelLink]:
    return session.scalar(
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.taxonomy_node_id == node_id)
        .where(MasterTaxonomyChannelLink.channel_code == channel_code)
        .where(MasterTaxonomyChannelLink.connection_id == connection_id)
        .where(MasterTaxonomyChannelLink.taxonomy_kind == taxonomy_kind)
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
        .limit(1)
    )


def _links_for_node(session: Session, node_id: int) -> Dict[str, MasterTaxonomyChannelLink]:
    rows = session.scalars(
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.taxonomy_node_id == node_id)
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
    ).all()
    return {row.channel_code: row for row in rows}


def _upsert_channel_link(session: Session, **payload: Any) -> MasterTaxonomyChannelLink:
    from db.master_taxonomy_hierarchy import _upsert_channel_link as upsert

    upsert(session, **payload)
    session.flush()
    link = _channel_link(
        session,
        int(payload["node_id"]),
        str(payload["channel_code"]),
        int(payload["connection_id"]),
        str(payload.get("taxonomy_kind") or MAGENTO_KIND),
    )
    if link is None:
        raise RuntimeError("channel link upsert failed")
    return link


def _write_listing_target(
    session: Session,
    node: MasterTaxonomyNode,
    channel_code: str,
    connection_id: int,
    remote_id: str,
    taxonomy_kind: str,
    remote_path: str,
) -> None:
    _ensure_listing_path_for_node(session, node)
    upsert_listing_path_target(
        session,
        path_slug=node.path_slug,
        channel_code=channel_code,
        remote_id=remote_id,
        taxonomy_kind=taxonomy_kind,
        connection_id=connection_id,
        remote_path=remote_path,
    )


def _upsert_magento_registry(
    session: Session,
    connection_id: int,
    category_id: int,
    parent_id: int,
    name: str,
    node: MasterTaxonomyNode,
) -> None:
    repo = SqlAlchemyCategoryRegistryRepository(session)
    repo.upsert_category(
        connection_id,
        {
            "category_id": category_id,
            "parent_id": parent_id,
            "name": name,
            "path": str(category_id),
            "path_names": node.name.replace(" / ", "/"),
            "level": node.path_slug.count("/") + 1,
            "is_active": True,
        },
        datetime.now(timezone.utc),
    )


def _upsert_shopify_registry(
    session: Session,
    connection_id: Optional[int],
    shop_code: Optional[str],
    collection_id: str,
    collection: Dict[str, Any],
    title: str,
) -> None:
    row = session.scalar(
        select(ShopifyCollectionRegistry)
        .where(ShopifyCollectionRegistry.collection_id == collection_id)
        .limit(1)
    )
    fetched_at = datetime.now(timezone.utc)
    if row is None:
        row = ShopifyCollectionRegistry(
            shop_code=shop_code or "default",
            connection_id=connection_id,
            collection_id=collection_id,
            fetched_at=fetched_at,
        )
        session.add(row)
    row.handle = collection.get("handle") or row.handle
    row.title = collection.get("title") or title
    row.collection_type = "manual"
    row.fetched_at = fetched_at
    session.flush()


def _parent_path_slug(session: Session, parent_id: Optional[int]) -> Optional[str]:
    if not parent_id:
        return None
    parent = session.get(MasterTaxonomyNode, int(parent_id))
    return parent.path_slug if parent else None


def _leaf_title(name: str) -> str:
    return str(name or "").strip().split(" / ")[-1].strip()


def _sync_node_kind_for_parent(node: MasterTaxonomyNode) -> None:
    """Hubs are root-only; nodes with a parent should be categories (collections unchanged)."""
    if node.node_kind == NODE_COLLECTION:
        return
    if node.parent_id:
        if node.node_kind == NODE_HUB:
            node.node_kind = NODE_CATEGORY
    elif node.node_kind == NODE_CATEGORY:
        node.node_kind = NODE_HUB


def _is_shopify_handle_taken_error(exc: BaseException) -> bool:
    text = str(exc).lower()
    return "handle has already been taken" in text


def _remote_linked_taxonomy_node(
    session: Session,
    channel_code: str,
    connection_id: int,
    remote_id: str,
    *,
    exclude_node_id: Optional[int] = None,
) -> Optional[MasterTaxonomyNode]:
    stmt = (
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.channel_code == channel_code)
        .where(MasterTaxonomyChannelLink.connection_id == connection_id)
        .where(MasterTaxonomyChannelLink.remote_id == remote_id)
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
    )
    if exclude_node_id is not None:
        stmt = stmt.where(MasterTaxonomyChannelLink.taxonomy_node_id != int(exclude_node_id))
    link = session.scalar(stmt.limit(1))
    if link is None:
        return None
    return session.get(MasterTaxonomyNode, link.taxonomy_node_id)


def _enrich_channel_candidate(
    session: Session,
    channel_code: str,
    connection_id: int,
    candidate: Dict[str, Any],
    *,
    exclude_node_id: Optional[int] = None,
) -> Dict[str, Any]:
    enriched = dict(candidate)
    other = _remote_linked_taxonomy_node(
        session,
        channel_code,
        connection_id,
        str(candidate.get("remote_id") or ""),
        exclude_node_id=exclude_node_id,
    )
    enriched["linked_taxonomy_node_id"] = other.id if other else None
    enriched["linked_taxonomy_node_name"] = other.name if other else None
    return enriched


def _link_shopify_collection_to_node(
    session: Session,
    node: MasterTaxonomyNode,
    *,
    connection_id: int,
    shop_code: Optional[str],
    collection_id: str,
    collection: Dict[str, Any],
    title: str,
) -> None:
    _upsert_shopify_registry(session, connection_id, shop_code, collection_id, collection, title)
    _upsert_channel_link(
        session,
        node_id=node.id,
        channel_code="shopify",
        connection_id=connection_id,
        taxonomy_kind=SHOPIFY_COLLECTION_KIND,
        remote_id=collection_id,
        remote_parent_id="0",
        remote_path=title,
    )
    _write_listing_target(
        session, node, "shopify", connection_id, collection_id, SHOPIFY_COLLECTION_KIND, title
    )


def _node_payload(
    node: MasterTaxonomyNode,
    links: Dict[str, MasterTaxonomyChannelLink],
    *,
    category_icon_url: Optional[str] = None,
) -> Dict[str, Any]:
    channels = {
        code: {
            "remote_id": link.remote_id,
            "remote_parent_id": link.remote_parent_id,
            "remote_path": link.remote_path,
            "taxonomy_kind": link.taxonomy_kind,
            "connection_id": link.connection_id,
        }
        for code, link in links.items()
    }
    payload = {
        "id": node.id,
        "parent_id": node.parent_id or 0,
        "node_kind": node.node_kind,
        "code": node.code,
        "name": node.name,
        "path_slug": node.path_slug,
        "collection_registry_id": node.collection_registry_id,
        "sort_order": node.sort_order,
        "channels": channels,
    }
    if node.node_kind in {NODE_HUB, NODE_CATEGORY}:
        payload["category_icon_url"] = category_icon_url
    return payload


def _upsert_taxonomy_category_icon(
    session: Session,
    node: MasterTaxonomyNode,
    icon_url: Any,
) -> None:
    """Persist Start Shopping icon URL for hub/category nodes (external CDN/R2 URL)."""
    from db.collection_landing_pages import DEFAULT_CATEGORY_L1

    if node.node_kind == NODE_COLLECTION:
        return
    slug = normalize_plp_path(node.path_slug)
    icon = str(icon_url or "").strip() or None
    row = session.scalar(select(CollectionLandingPage).where(CollectionLandingPage.path_slug == slug).limit(1))
    if row is None:
        if not icon:
            return
        row = CollectionLandingPage(
            path_slug=slug,
            page_type="category",
            category_l1=DEFAULT_CATEGORY_L1,
            title=node.name,
            collection=node.name,
            category_icon_url=icon,
            is_active=True,
        )
        session.add(row)
    else:
        row.category_icon_url = icon
    session.flush()


def _push_taxonomy_category_icon_magento(
    session: Session,
    api: Any,
    node: MasterTaxonomyNode,
    category_id: int,
    *,
    dry_run: bool,
) -> None:
    if node.node_kind == NODE_COLLECTION or api is None:
        return
    landing = session.scalar(
        select(CollectionLandingPage).where(CollectionLandingPage.path_slug == node.path_slug).limit(1)
    )
    icon_url = str(landing.category_icon_url or "").strip() if landing else ""
    if not icon_url:
        return
    from magento.collection_landing_push import push_magento_collection_landing

    push_magento_collection_landing(
        api,
        category_id=int(category_id),
        collection_row={"category_icon_url": icon_url},
        field_keys={"category_icon_url"},
        dry_run=dry_run,
    )


def link_products_to_taxonomy_from_master(
    session: Session,
    *,
    skus: Optional[Sequence[str]] = None,
    only_unlinked: bool = False,
    replace_existing: bool = False,
    sync_channels: bool = False,
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    dry_run: bool = True,
    limit: Optional[int] = None,
) -> Dict[str, Any]:
    """Assign master SKUs to taxonomy paths (collection + shopping facet + detail category).

    Includes Start Shopping L1 facet ancestors (e.g. wall-cabinets) so Magento layered
    category filters on collection pages can AND collection ∩ facet membership.
    """
    from db.channel_listing_path import bulk_assign_listing_paths, resolve_listing_paths_for_sku
    from db.master_taxonomy_hierarchy import _attrs_by_sku
    from db.master_taxonomy_product_paths import canonical_master_taxonomy_path_slugs
    from db.models import ProductChannelTaxonomyAssignment, ProductListingPathAssignment
    from db.product_channel_taxonomy import (
        MAGENTO_CATEGORY_KINDS,
        bulk_assign_sku_taxonomy,
        SHOPIFY_COLLECTION_KIND,
        SHOPIFY_PRODUCT_CATEGORY_KIND,
    )

    stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True))
    if skus:
        stmt = stmt.where(MasterProduct.sku.in_([str(s).strip() for s in skus if str(s).strip()]))
    if limit:
        stmt = stmt.limit(int(limit))
    products = list(session.scalars(stmt).all())
    attrs_by_sku = _attrs_by_sku(session, [p.id for p in products])

    assignments: Dict[str, set[str]] = {}
    per_sku_slugs: Dict[str, List[str]] = {}
    skipped_no_paths = 0
    skipped_already_linked = 0

    for product in products:
        if only_unlinked and resolve_listing_paths_for_sku(session, product.sku):
            skipped_already_linked += 1
            continue
        slugs = canonical_master_taxonomy_path_slugs(
            session,
            product,
            attrs_by_sku.get(product.sku, {}),
        )
        if not slugs:
            skipped_no_paths += 1
            continue
        per_sku_slugs[product.sku] = slugs
        for slug in slugs:
            assignments.setdefault(slug, set()).add(product.sku)

    assignment_count = sum(len(sku_set) for sku_set in assignments.values())
    stats: Dict[str, Any] = {
        "dry_run": dry_run,
        "product_count": len(products),
        "path_count": len(assignments),
        "assignment_count": assignment_count,
        "skipped_no_paths": skipped_no_paths,
        "skipped_already_linked": skipped_already_linked,
        "replace_existing": replace_existing,
        "sync_channels": sync_channels,
        "deactivated_listing_paths": 0,
        "deactivated_channel_assignments": 0,
        "would_deactivate_listing_paths": 0,
        "would_deactivate_channel_assignments": 0,
        "missing_channel_links": {"magento": 0, "shopify": 0},
        "samples": [
            {"path_slug": slug, "sku_count": len(sku_set), "skus_sample": sorted(sku_set)[:5]}
            for slug, sku_set in sorted(assignments.items())[:20]
        ],
    }

    affected_skus = sorted(per_sku_slugs)
    if affected_skus:
        stats["would_deactivate_listing_paths"] = len(
            list(
                session.scalars(
                    select(ProductListingPathAssignment).where(
                        ProductListingPathAssignment.master_sku.in_(affected_skus),
                        ProductListingPathAssignment.assignment_status == ACTIVE_STATUS,
                    )
                ).all()
            )
        )
        active_channel_rows = list(
            session.scalars(
                select(ProductChannelTaxonomyAssignment).where(
                    ProductChannelTaxonomyAssignment.master_sku.in_(affected_skus),
                    ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS,
                )
            ).all()
        )
        stats["would_deactivate_channel_assignments"] = sum(
            1
            for row in active_channel_rows
            if _should_replace_channel_taxonomy_assignment(
                row,
                magento_connection_id=magento_connection_id,
                shopify_connection_id=shopify_connection_id,
                magento_category_kinds=MAGENTO_CATEGORY_KINDS,
                shopify_product_category_kind=SHOPIFY_PRODUCT_CATEGORY_KIND,
            )
        )
    if dry_run or not assignments:
        return stats

    if replace_existing:
        for product in products:
            if product.sku not in per_sku_slugs:
                continue
            for row in session.scalars(
                select(ProductListingPathAssignment).where(ProductListingPathAssignment.master_sku == product.sku)
            ).all():
                if row.assignment_status == ACTIVE_STATUS:
                    row.assignment_status = INACTIVE_STATUS
                    stats["deactivated_listing_paths"] += 1
            for row in session.scalars(
                select(ProductChannelTaxonomyAssignment).where(
                    ProductChannelTaxonomyAssignment.master_sku == product.sku,
                    ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS,
                )
            ).all():
                if not _should_replace_channel_taxonomy_assignment(
                    row,
                    magento_connection_id=magento_connection_id,
                    shopify_connection_id=shopify_connection_id,
                    magento_category_kinds=MAGENTO_CATEGORY_KINDS,
                    shopify_product_category_kind=SHOPIFY_PRODUCT_CATEGORY_KIND,
                ):
                    continue
                row.assignment_status = INACTIVE_STATUS
                stats["deactivated_channel_assignments"] += 1
        session.flush()

    # Ensure listing paths exist for every assigned slug (all-products has no
    # taxonomy node; collection/category paths may pre-exist from hierarchy sync).
    from db.master_taxonomy_product_paths import all_products_path_slug
    from db.master_taxonomy_intent import _NodeIndex

    root_slug = all_products_path_slug()
    index = _NodeIndex.load(session)
    for path_slug in assignments:
        if path_slug == root_slug:
            upsert_listing_path(
                session,
                path_slug=path_slug,
                title=ALL_PRODUCTS,
                path_kind="category",
                notes="global Magento Default/Root + Shopify All membership",
            )
            continue
        node = index.by_slug.get(normalize_plp_path(path_slug))
        title = node.name if node is not None else path_slug
        kind = node.node_kind if node is not None else None
        upsert_listing_path(
            session,
            path_slug=path_slug,
            title=title,
            path_kind=kind,
            notes="linked by master taxonomy product reconcile",
        )
    session.flush()

    upserted = 0
    for path_slug, sku_set in assignments.items():
        result = bulk_assign_listing_paths(
            session,
            path_slugs=[path_slug],
            skus=sorted(sku_set),
            dry_run=False,
            replace_existing=False,
        )
        upserted += int(result.get("upserted") or result.get("assignment_count") or 0)
    stats["upserted"] = upserted

    if sync_channels:
        channel_stats = _sync_channel_taxonomy_for_sku_paths(
            session,
            per_sku_slugs,
            magento_connection_id=magento_connection_id,
            shopify_connection_id=shopify_connection_id,
        )
        root_stats = _assign_global_root_channel_memberships(
            session,
            sorted(per_sku_slugs),
            magento_connection_id=magento_connection_id,
            shopify_connection_id=shopify_connection_id,
        )
        channel_stats = {**channel_stats, "global_root": root_stats}
        stats["channel_taxonomy"] = channel_stats
        stats["missing_channel_links"] = channel_stats.get("missing_channel_links", stats["missing_channel_links"])

    return stats


def _should_replace_channel_taxonomy_assignment(
    row: ProductChannelTaxonomyAssignment,
    *,
    magento_connection_id: Optional[int],
    shopify_connection_id: Optional[int],
    magento_category_kinds: set[str],
    shopify_product_category_kind: str,
) -> bool:
    channel = str(row.channel_code or "").strip().lower()
    if channel == "magento":
        if magento_connection_id is None:
            return False
        if row.taxonomy_kind not in magento_category_kinds:
            return False
        return row.connection_id is None or int(row.connection_id) == int(magento_connection_id)
    if channel == "shopify":
        if shopify_connection_id is None:
            return False
        # Shopify's standard product category is a separate concern from
        # our master taxonomy -> collection reconciliation.
        if row.taxonomy_kind == shopify_product_category_kind:
            return False
        return row.connection_id is None or int(row.connection_id) == int(shopify_connection_id)
    return False


def _sync_channel_taxonomy_for_sku_paths(
    session: Session,
    per_sku_slugs: Dict[str, List[str]],
    *,
    magento_connection_id: Optional[int],
    shopify_connection_id: Optional[int],
) -> Dict[str, Any]:
    from db.master_taxonomy_product_paths import _NodeIndex, all_products_path_slug

    index = _NodeIndex.load(session)
    magento_count = 0
    shopify_count = 0
    missing_magento = 0
    missing_shopify = 0
    root_slug = all_products_path_slug()

    for master_sku, slugs in per_sku_slugs.items():
        for slug in slugs:
            normalized = normalize_plp_path(slug)
            # Handled by _assign_global_root_channel_memberships (no taxonomy node).
            if normalized == root_slug:
                continue
            node = index.by_slug.get(normalized)
            if node is None:
                continue
            links = _links_for_node(session, node.id)
            magento_link = links.get("magento")
            if (
                magento_connection_id
                and magento_link
                and magento_link.remote_id
                and (
                    magento_link.connection_id is None
                    or int(magento_link.connection_id) == int(magento_connection_id)
                )
            ):
                bulk_assign_sku_taxonomy(
                    session,
                    channel_code="magento",
                    remote_ids=[magento_link.remote_id],
                    taxonomy_kind=magento_link.taxonomy_kind or MAGENTO_KIND,
                    connection_id=magento_connection_id,
                    skus=[master_sku],
                    remote_path=magento_link.remote_path or node.name,
                    replace_existing=False,
                    dry_run=False,
                )
                magento_count += 1
            elif magento_connection_id:
                missing_magento += 1
            shopify_link = links.get("shopify")
            if (
                shopify_connection_id
                and shopify_link
                and shopify_link.remote_id
                and (
                    shopify_link.connection_id is None
                    or int(shopify_link.connection_id) == int(shopify_connection_id)
                )
            ):
                bulk_assign_sku_taxonomy(
                    session,
                    channel_code="shopify",
                    remote_ids=[shopify_link.remote_id],
                    taxonomy_kind=SHOPIFY_COLLECTION_KIND,
                    connection_id=shopify_connection_id,
                    skus=[master_sku],
                    remote_path=shopify_link.remote_path or node.name,
                    replace_existing=False,
                    dry_run=False,
                )
                shopify_count += 1
            elif shopify_connection_id:
                missing_shopify += 1

    return {
        "magento_assignments": magento_count,
        "shopify_assignments": shopify_count,
        "missing_channel_links": {
            "magento": missing_magento,
            "shopify": missing_shopify,
        },
    }


def _assign_global_root_channel_memberships(
    session: Session,
    skus: Sequence[str],
    *,
    magento_connection_id: Optional[int],
    shopify_connection_id: Optional[int],
) -> Dict[str, Any]:
    """Assign Magento Default/Root Category and Shopify All collection to SKUs.

    These remotes are not master_taxonomy_nodes (hierarchy skips All Products), so
    they are resolved from channel registries and linked to the ``all-products``
    listing path.
    """
    from db.master_taxonomy_product_paths import all_products_path_slug

    master_skus = [str(s).strip() for s in skus if str(s).strip()]
    root_slug = all_products_path_slug()
    stats: Dict[str, Any] = {
        "path_slug": root_slug,
        "sku_count": len(master_skus),
        "magento_remote_id": None,
        "magento_remote_path": None,
        "magento_assigned": 0,
        "shopify_remote_id": None,
        "shopify_remote_path": None,
        "shopify_assigned": 0,
        "magento_missing": False,
        "shopify_missing": False,
    }
    if not master_skus or not root_slug:
        return stats

    upsert_listing_path(
        session,
        path_slug=root_slug,
        title=ALL_PRODUCTS,
        path_kind="category",
        notes="global Magento Default/Root + Shopify All membership",
    )

    if magento_connection_id is not None:
        magento_remote = _resolve_magento_default_root_category(session, int(magento_connection_id))
        if magento_remote is None:
            stats["magento_missing"] = True
        else:
            category_id, remote_path = magento_remote
            stats["magento_remote_id"] = str(category_id)
            stats["magento_remote_path"] = remote_path
            upsert_listing_path_target(
                session,
                path_slug=root_slug,
                channel_code="magento",
                remote_id=str(category_id),
                taxonomy_kind=MAGENTO_KIND,
                connection_id=int(magento_connection_id),
                remote_path=remote_path,
            )
            result = bulk_assign_sku_taxonomy(
                session,
                channel_code="magento",
                remote_ids=[str(category_id)],
                taxonomy_kind=MAGENTO_KIND,
                connection_id=int(magento_connection_id),
                skus=master_skus,
                remote_path=remote_path,
                only_assigned=True,
                replace_existing=False,
                dry_run=False,
            )
            stats["magento_assigned"] = int(result.get("upserted") or result.get("assignment_count") or len(master_skus))

    if shopify_connection_id is not None:
        shopify_remote = _resolve_shopify_all_collection(session, int(shopify_connection_id))
        if shopify_remote is None:
            stats["shopify_missing"] = True
        else:
            collection_id, remote_title = shopify_remote
            stats["shopify_remote_id"] = collection_id
            stats["shopify_remote_path"] = remote_title
            upsert_listing_path_target(
                session,
                path_slug=root_slug,
                channel_code="shopify",
                remote_id=collection_id,
                taxonomy_kind=SHOPIFY_COLLECTION_KIND,
                connection_id=int(shopify_connection_id),
                remote_path=remote_title,
            )
            result = bulk_assign_sku_taxonomy(
                session,
                channel_code="shopify",
                remote_ids=[collection_id],
                taxonomy_kind=SHOPIFY_COLLECTION_KIND,
                connection_id=int(shopify_connection_id),
                skus=master_skus,
                remote_path=remote_title,
                only_assigned=True,
                replace_existing=False,
                dry_run=False,
            )
            stats["shopify_assigned"] = int(result.get("upserted") or result.get("assignment_count") or len(master_skus))

    return stats


def _resolve_magento_default_root_category(
    session: Session,
    connection_id: int,
) -> Optional[tuple[int, str]]:
    """Resolve Magento Default/Root (fallback All Products) from the category registry."""
    from datetime import timedelta

    repo = SqlAlchemyCategoryRegistryRepository(session)
    ttl_cutoff = datetime.now(timezone.utc) - timedelta(days=3650)
    for path_names in (
        "Default Category",
        "Root Catalog",
        "All Products",
        "Default Category/All Products",
    ):
        category_id = repo.resolve_path(connection_id, path_names, ttl_cutoff)
        if category_id:
            return int(category_id), path_names
    for name in ("Default Category", "Root Catalog", "All Products"):
        category_id = repo.resolve_name(connection_id, name, ttl_cutoff)
        if category_id:
            return int(category_id), name
    # Hard fallback used elsewhere when creating Magento category trees.
    row = session.scalar(
        select(MagentoCategoryRegistry).where(
            MagentoCategoryRegistry.connection_id == connection_id,
            MagentoCategoryRegistry.category_id == MAGENTO_ROOT_CATEGORY_ID,
            MagentoCategoryRegistry.is_active.is_(True),
        )
    )
    if row is not None:
        return int(row.category_id), str(row.path_names or row.name or "Default Category")
    return None


def _resolve_shopify_all_collection(
    session: Session,
    connection_id: int,
) -> Optional[tuple[str, str]]:
    """Resolve Shopify All / All Products collection from the registry."""
    rows = list(
        session.scalars(
            select(ShopifyCollectionRegistry).where(
                (ShopifyCollectionRegistry.connection_id == connection_id)
                | (ShopifyCollectionRegistry.connection_id.is_(None))
            )
        ).all()
    )
    if not rows:
        return None

    def _norm(value: Optional[str]) -> str:
        return str(value or "").strip().lower()

    preferred_titles = {"all", "all products"}
    preferred_handles = {"all", "all-products"}
    for row in rows:
        if _norm(row.title) in preferred_titles or _norm(row.handle) in preferred_handles:
            collection_id = str(row.collection_id or "").strip()
            if collection_id:
                return collection_id, str(row.title or row.handle or "All")
    return None