PhoenixKitEcommerce.Workers.ShopifyMediaSyncWorker (PhoenixKitEcommerce v0.5.0)

Copy Markdown View Source

Oban worker driving the three Block 7 catalogue writers as background jobs — Task 5 of docs/superpowers/plans/2026-09-06-block7-shopify- media-collections.md.

Job arguments

%{"kind" => "images" | "variants" | "collections", "actor_uuid" => uuid_or_nil}kind selects which writer runs; actor_uuid is the Storage file owner for "images" (ignored by the other two kinds, carried uniformly anyway so the sync page never has to special-case the enqueue call per button).

Products -> items: how "images"/"variants" find their item

Both kinds fetch every Shopify product once (opts[:client], default AdminClient.fetch_products/2) and, for each one, look up the catalogue item it belongs to: first by data["ecommerce"]["shopify"] ["product_id"] (stringified), then — because product_id is only backfilled onto an item the first time a regular field sync applies a change to it (Writer.update_from_shopify/3; see Task 1), so plenty of items carry only handle until that has happened — by data ["ecommerce"]["shopify"]["handle"]. A product matching neither is recorded as an error ("no_matching_item") rather than skipped silently.

Each product is independent: a Writer.sync_images/3 or sync_variants/2 failure on one product is recorded in the run's errors list and the loop moves on to the next product — one bad product must not stop the other ~664. This is CSVImportWorker's own per-row philosophy, not CollectionSync's (which halts on a write failure because collection membership assignment is one connected pass, not independent rows).

"collections"

Delegates entirely to PhoenixKitEcommerce.Shopify.CollectionSync.run/1 — a single unit of work (total: 1), whose own {:ok, stats} map is kept as the progress record's "result" for the sync page to display ("last result"); actor_uuid plays no part here.

Progress record

One phoenix_kit_shop_config row, key "shopify_media_sync" (deliberately singular — Task 7 runs the three kinds one at a time, never concurrently, so a single record naming its own "kind" is enough to know what it describes and whether that specific button should show as in-flight):

%{"kind" => "images" | "variants" | "collections",
  "total" => non_neg_integer(), "done" => non_neg_integer(),
  "errors" => [%{"product" => String.t(), "reason" => String.t()}],
  "started_at" => iso8601, "finished_at" => iso8601 | nil,
  "result" => map() | nil}

A job in flight has "finished_at" => nil; a caller reading this to decide whether to disable a button matches progress["kind"] against the button's own kind first. Every write also broadcasts on topic/0 (Manager.broadcast/2) so the sync page's LiveView can update live instead of polling — mirrors CSVImportWorker's own shop:import:* broadcasts.

A no-op — {:error, :catalogue_source_inactive} — when ProductSource.current/0 isn't Catalogue (checked here too, even though every writer this dispatches to already self-gates: fetching Shopify products and building the item index first would be wasted work under the legacy source).

Summary

Functions

Reads the current (or last) progress record, nil if none exists yet.

Runs one sync kind directly — what perform/1 calls with production defaults. opts

The PubSub topic the sync page subscribes to for live progress.

Functions

get_progress()

@spec get_progress() :: map() | nil

Reads the current (or last) progress record, nil if none exists yet.

run(kind, actor_uuid, opts \\ [])

@spec run(String.t(), String.t() | nil, keyword()) ::
  {:ok, map()} | {:error, :catalogue_source_inactive | term()}

Runs one sync kind directly — what perform/1 calls with production defaults. opts:

  • :client — module with fetch_products/2 (images/variants) and/or fetch_collections/1/fetch_collection_product_ids/2 (collections); defaults to AdminClient.
  • :downloader — forwarded to Writer.sync_images/3's own opts[:downloader].
  • :integration_uuid — skips resolving the shop's Shopify connection (tests inject this; production always resolves it).

Exists as a public function, separate from perform/1, so tests can exercise the real logic without going through Oban/HTTP.

topic()

@spec topic() :: String.t()

The PubSub topic the sync page subscribes to for live progress.