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
@spec get_progress() :: map() | nil
Reads the current (or last) progress record, nil if none exists yet.
@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 withfetch_products/2(images/variants) and/orfetch_collections/1/fetch_collection_product_ids/2(collections); defaults toAdminClient.:downloader— forwarded toWriter.sync_images/3's ownopts[: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.
@spec topic() :: String.t()
The PubSub topic the sync page subscribes to for live progress.