@spec create_from_shopify(map(), String.t()) ::
  {:ok, PhoenixKitCatalogue.Schemas.Item.t()}
  | {:error, Ecto.Changeset.t() | [{atom(), String.t()}] | :catalogue_not_found}

Creates a catalogue item from a Shopify Admin API product payload for a handle with no local match (ProductDiff.new_product_changes/3).

name/description are written as the item's own columns (a brand new item has no other language yet, so base_locale — whatever locale the sync ran in — IS this item's primary language); slug[base_locale] comes from Slugs.from_title/3, retried with a -2/-3 numeric suffix on a slug collision (3 attempts total, same shape the data migration's own slug retry uses); base_price from the cheapest variant; markup_percentage 0 (Shopify price is the single source of truth — see the design spec's pricing principle); unit "piece"; status "active"; category_uuid nil (uncategorized, same as a legacy-sync-created product used to be — sorting into a category is a manual follow-up either way); data["ecommerce"] carries shopify: %{"handle" => ..., "product_id" => ...} (product_id stringified, same as update_from_shopify/3's own backfill — every reader, CollectionSync and the Task 5 worker's item index included, matches it as a string) and shop_status derived from the Shopify product's own status.

@spec sync_images(PhoenixKitCatalogue.Schemas.Item.t(), map(), keyword()) ::
  {:ok,
   %{
     downloaded: non_neg_integer(),
     reused: non_neg_integer(),
     attached: non_neg_integer(),
     errors: [{String.t(), term()}]
   }}
  | {:error, :catalogue_source_inactive | term()}

Downloads shopify_product's "images" into Storage and attaches them to item in Shopify's own position order (ascending; the payload's own array order is NOT trusted), featured = the position-1 image.

Resolves each Shopify image in this order: (a) an id already in data ["ecommerce"]["shopify"]["image_ids"] (%{"<shopify image id>" => file_uuid}) reuses its file uuid; (b) failing that, a Storage file already linked to item — one of its data["media_order"]/ featured_image_uuid files, or any file living in (or folder-linked into) its data["files_folder_uuid"] — whose metadata["source_url"] matches the Shopify image's src once both are stripped of their ?v=-style query string reuses that file instead of downloading a second copy of it (this is what lets a rerun against a catalogue item migrated with plain Storage images, never tagged with a Shopify image id, converge without duplicating every file); (c) otherwise downloads via opts[:downloader] (default &ImageDownloader.download_and_store/ 3, (url, user_uuid, opts) -> {:ok, file_uuid} | {:error, reason}).

opts[:user_uuid] is the Storage file owner for anything downloaded; when omitted it falls back to PhoenixKit.Users.Auth. get_first_admin_uuid/0 so an unattended sync run never fails Storage's user_uuid can't be blank validation for lack of an explicit actor.

A download failure never corrupts a binding that does not belong to it: an image id already synced before (image_ids) or matched to a previously-linked file by source_url is resolved WITHOUT ever attempting a download (see (a)/(b) above), so an existing binding can never be lost to a transient failure. An image with neither — a genuinely new Shopify image whose download fails — is skipped: not attached, not recorded in image_ids, so a later run retries it. It is never bound to some other image's file merely because it landed at the same list position; still listed in :errors so the failure stays visible.

A no-op — {:error, :catalogue_source_inactive} — when ProductSource.current/0 isn't Catalogue (Global Constraints: every new Block 7 writer is legacy-source-safe on its own).

Idempotent: a second run against the same payload resolves every image id to its already-known file uuid (downloaded: 0) and re-attaches the same order. image_ids is rewritten fresh from the current sync each run (same "derived fresh" idiom sync_variants/2 uses for price_modifiers/set_slugs) — an id Shopify no longer lists is dropped from it, though the file itself is left attached (no deletions in this block).

@spec sync_variants(PhoenixKitCatalogue.Schemas.Item.t(), map()) ::
  {:ok, %{sets: non_neg_integer(), values_created: non_neg_integer()}}
  | {:error, :catalogue_source_inactive | term()}

Turns shopify_product's options/variants (PhoenixKitEcommerce.Shopify.VariantMapper.build/1) into catalogue attribute-set attachments on item: one set per real Shopify option (found by blueprint name "catalogue_set_" <> slug, created kind: "fixed" when missing), values resolved to slugs via ValueResolver. resolve_many/3 (unknown labels become draft values), attached in Shopify's option order and selected in label order, with a slug-keyed price-modifier map written to data["ecommerce"] ["price_modifiers"][set_slug].

A no-op — {:error, :catalogue_source_inactive} — when ProductSource.current/0 isn't Catalogue (Global Constraints: every new Block 7 writer is legacy-source-safe on its own, not only via whatever caller happens to gate it).

Idempotent: a second call against the same shopify_product resolves every label to its already-created slug (values_created: 0), leaves already-selected/attached sets untouched (no write, no activity row — AttributeSets.attach_set/3 and set_attachment_selection/4 are both no-ops on an unchanged state), and rewrites the same price_modifiers/ set_slugs. Any set previously written by a Shopify sync (tracked in data["ecommerce"]["shopify"]["set_slugs"]) that this product no longer has options for is detached and dropped from both that list and price_modifiers.

@spec update_from_shopify(PhoenixKitCatalogue.Schemas.Item.t(), map(), String.t()) ::
  {:ok, PhoenixKitCatalogue.Schemas.Item.t()}
  | {:error, Ecto.Changeset.t() | [{atom(), String.t()}]}

Applies change_fields — a plain %{field_atom => incoming_value} map, built by Shopify.Sync.apply_change/2 from a ProductDiff.Change's changes (unwrapped of its %{current:, incoming:} shape) — to item, writing localized fields into base_locale (the SAME locale the change was diffed against — see ProductDiff.Change's moduledoc for why that matters).

Recognized keys: :title, :body_html, :description, :vendor, :tags, :status (mapped to data["ecommerce"]["shop_status"]), :price (→ base_price), :compare_at_price. Any other key is ignored — this mirrors ProductDiff.comparable_fields/0's set, but doesn't hard-code it, so a caller that already filtered change_fields (e.g. to a single field an operator picked) never has to know that.

:handle and :product_id are the exception: not part of ProductDiff.comparable_fields/0, they are merged into data["ecommerce"]["shopify"] (product_id stringified) whenever present, alongside whatever Writer or a later Shopify sync already wrote there (image_ids, set_slugs, collection_id) — never replacing that sub-map wholesale. Shopify.Sync.apply_change/2 sets both on every applied Change, backfilling identity even when the caller only asked for a subset of the diffed fields.