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).