Reconciler Storefront Visibility
Active assigned products still absent from storefront listings
The product is active, it is assigned to the sales channel, its product_visibility row says visibility: 30, and it sits in the right category. Pull it with GET /api/product/{id} and everything checks out. Yet the storefront listing and search pages act like it does not exist. Here is why the storefront's read side can fall behind the true state and a small script that finds the exact products affected and forces them back in sync.
Shopware's storefront listing and search never query the live product table directly. They read a denormalized, asynchronously built view (the DAL indexes plus, if enabled, Elasticsearch) that is kept current by product.indexer messages consumed through messenger:consume. A product must have active=true, a sales channel assignment, a product_visibility row with visibility=30 for that channel, and correct category linkage, but if the write that set those fields correctly never got its indexing message consumed, the read side simply never learned about it. The product is fully correct in the database and in GET /api/product/{id}, yet sales-channel.product.repository still returns nothing for it. Run a small Python or Node.js script that pulls active products with their visibilities and categories, checks whether the required product_visibility row is missing or wrong, confirms the category linkage, and checks whether the product.indexer scheduled task's lastExecutionTime is stale relative to the product's last write. In dry run it only reports what it finds. Only when told to write does it trigger a scoped reindex, or, in the rarer case the visibility row itself is missing or wrong, create or fix that one row. Full code, tests, and sources are below.
The problem in plain words
It is easy to assume the Admin and the storefront are looking at the same thing. They are not. The Admin API reads the live, normalized product table straight from the database. The storefront listing and search routes read a separate, denormalized view that Shopware builds and keeps warm in the background, through the DAL indexers and, if Elasticsearch is turned on, through Shopware\Elasticsearch\Product\ProductUpdater. That view only updates when the matching indexing message is actually processed.
So there are really two questions to ask about a missing product, not one. First, is the underlying data correct: active, assigned, visible, categorized? Second, has the storefront's read side actually caught up with that data? Most troubleshooting stops at the first question, sees everything looking fine, and concludes the bug must be somewhere else. The real gap is almost always the second question: the write went through the Data Abstraction Layer just fine, but the corresponding EntityIndexingMessage either never got queued, or got queued and never got consumed because messenger:consume stalled or was not running.
Why it happens
This is a write side versus read side gap, not a data problem, most of the time. A few common ways stores end up here:
- A bulk import or API write goes through the DAL correctly, but the worker running
messenger:consumecrashed, was never started after a deploy, or was starved by a backlog of higher-priority messages, so theproduct.indexerorelasticsearch.indexingmessage for that product sits queued and unprocessed. - A category was reassigned through the Admin or an API script, and the category tree side of the index was rebuilt, but the product-to-category join used by the listing query was cached from before the change.
- A
product_visibilityrow was left atvisibility: 10or20after an import or a partial API write, so the product is technically visible in the database sense but hidden from listing, search, or both for that sales channel. - The scheduled task worker (
scheduled-task:run) is running, but no separate long-runningmessenger:consumeworker is, so scheduled tasks fire on time while the actual entity indexing messages they enqueue never get drained.
None of this throws a visible error to a merchant. The product passes every check in the Admin: active, in stock, assigned, visible. The only place the gap shows up is the storefront itself, where the product is simply absent, and there is nothing in the Admin UI that points back at a stale index or a stuck queue as the reason.
Being correct in the database and being visible in the storefront are two different claims, connected by a background job that has to actually run. So the fix is almost never a data write. It is confirming the underlying rows are already right, then forcing the indexer to reprocess the affected products and, separately, confirming the message consumer is actually draining what gets queued. Only in the rarer case where detection finds a genuinely missing or wrong product_visibility row do we touch that row directly, and even then we never touch active or any state machine field, since those are not the fault here.
The fix, as a flow
We do not guess which products are affected. The script pulls active products with their visibilities and categories, checks each one against the storefront's requirements with one pure function, and only for products that are already correctly configured but whose index looks stale does it trigger a reindex. If it instead finds a missing or wrong visibility row, it repairs that row directly, never anything else.
Build it step by step
Get Admin API credentials
Create an Integration in your Shopware Administration under Settings, System, Integrations. Copy the access key id and secret access key. The script exchanges these for a bearer token at POST /api/oauth/token with a client_credentials grant, so keep them in environment variables, never in the file.
pip install requests
export SHOPWARE_URL="https://your-shop.example.com"
export SHOPWARE_CLIENT_ID="SWIAxxx"
export SHOPWARE_CLIENT_SECRET="your secret access key"
export STOREFRONT_SALES_CHANNEL_ID="a1b2c3..."
export REQUIRED_CATEGORY_IDS="d4e5f6..."
export DRY_RUN="true" # start safe, change to false to write// Node 18+ has fetch built in, no dependencies needed
export SHOPWARE_URL="https://your-shop.example.com"
export SHOPWARE_CLIENT_ID="SWIAxxx"
export SHOPWARE_CLIENT_SECRET="your secret access key"
export STOREFRONT_SALES_CHANNEL_ID="a1b2c3..."
export REQUIRED_CATEGORY_IDS="d4e5f6..."
export DRY_RUN="true" // start safe, change to false to writeAuthenticate and talk to the Admin API
A small helper trades the client id and secret for an access token, then sends it as Authorization: Bearer <token> on every call. We reuse this helper for search, for reading the scheduled task, and for the write actions.
import os, requests
SHOPWARE_URL = os.environ["SHOPWARE_URL"].rstrip("/")
CLIENT_ID = os.environ["SHOPWARE_CLIENT_ID"]
CLIENT_SECRET = os.environ["SHOPWARE_CLIENT_SECRET"]
def get_token():
r = requests.post(
f"{SHOPWARE_URL}/api/oauth/token",
json={
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
},
headers={"Accept": "application/json"},
timeout=30,
)
r.raise_for_status()
return r.json()["access_token"]
def api(method, path, token, json_body=None):
r = requests.request(
method,
f"{SHOPWARE_URL}{path}",
json=json_body,
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json",
"Content-Type": "application/json",
},
timeout=30,
)
r.raise_for_status()
return r.json() if r.text else {}const SHOPWARE_URL = (process.env.SHOPWARE_URL || "").replace(/\/$/, "");
const CLIENT_ID = process.env.SHOPWARE_CLIENT_ID;
const CLIENT_SECRET = process.env.SHOPWARE_CLIENT_SECRET;
async function getToken() {
const res = await fetch(`${SHOPWARE_URL}/api/oauth/token`, {
method: "POST",
headers: { "Content-Type": "application/json", Accept: "application/json" },
body: JSON.stringify({
grant_type: "client_credentials",
client_id: CLIENT_ID,
client_secret: CLIENT_SECRET,
}),
});
if (!res.ok) throw new Error(`Shopware auth ${res.status}`);
const body = await res.json();
return body.access_token;
}
async function api(method, path, token, jsonBody) {
const res = await fetch(`${SHOPWARE_URL}${path}`, {
method,
headers: {
Authorization: `Bearer ${token}`,
Accept: "application/json",
"Content-Type": "application/json",
},
body: jsonBody ? JSON.stringify(jsonBody) : undefined,
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const text = await res.text();
return text ? JSON.parse(text) : {};
}Pull active products with visibilities and categories
Search product for active=true, with the visibilities and categories associations included, so each product carries exactly what the decision function needs. Page through with page and limit, reading total-count-mode:1 to know when to stop.
def search_active_products(token, page=1, limit=500):
body = {
"page": page,
"limit": limit,
"filter": [{"type": "equals", "field": "active", "value": True}],
"associations": {"visibilities": {}, "categories": {}},
"total-count-mode": 1,
}
return api("POST", "/api/search/product", token, body)async function searchActiveProducts(token, page = 1, limit = 500) {
const body = {
page,
limit,
filter: [{ type: "equals", field: "active", value: true }],
associations: { visibilities: {}, categories: {} },
"total-count-mode": 1,
};
return api("POST", "/api/search/product", token, body);
}Decide, with one pure function
Keep the decision in its own function that takes one product's already-fetched shape, the target sales channel id, the required category ids, the indexer's lastExecutionTime, and the product's last write time, and returns exactly one action. It never touches the network. It just reasons over inputs you already fetched, which makes it trivial to unit test with fixed dates.
def decide_visibility_repair(product, target_sales_channel_id, required_category_ids,
indexer_last_run, last_write_at):
if not product.get("active", False):
return {"action": "none", "reason": "inactive, not a listing bug"}
row = next(
(v for v in (product.get("visibilities") or [])
if v.get("salesChannelId") == target_sales_channel_id),
None,
)
if row is None:
return {"action": "create_visibility", "reason": "no product_visibility row for sales channel"}
if row.get("visibility") != 30:
return {"action": "fix_visibility", "reason": "visibility below 30 (hidden from listing/search)"}
category_ids = set(product.get("categoryIds") or [])
if not set(required_category_ids).issubset(category_ids):
return {"action": "flag_category_mismatch", "reason": "not linked to storefront category/stream"}
if indexer_last_run < last_write_at:
return {"action": "reindex", "reason": "index stale relative to last DB write"}
return {"action": "none", "reason": "configuration already correct"}export function decideVisibilityRepair(product, targetSalesChannelId, requiredCategoryIds,
indexerLastRun, lastWriteAt) {
if (!product.active) {
return { action: "none", reason: "inactive, not a listing bug" };
}
const row = (product.visibilities || []).find(
(v) => v.salesChannelId === targetSalesChannelId
);
if (!row) {
return { action: "create_visibility", reason: "no product_visibility row for sales channel" };
}
if (row.visibility !== 30) {
return { action: "fix_visibility", reason: "visibility below 30 (hidden from listing/search)" };
}
const categoryIds = new Set(product.categoryIds || []);
const missingCategory = requiredCategoryIds.some((id) => !categoryIds.has(id));
if (missingCategory) {
return { action: "flag_category_mismatch", reason: "not linked to storefront category/stream" };
}
if (indexerLastRun < lastWriteAt) {
return { action: "reindex", reason: "index stale relative to last DB write" };
}
return { action: "none", reason: "configuration already correct" };
}Read the indexer's own timestamp before deciding
Search scheduled-task for name=product.indexer and read back lastExecutionTime. Comparing that against when the product was last written is what actually tells you the queue is stuck, rather than guessing from the symptom alone.
import datetime
def parse_iso(value):
return datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
def get_product_indexer_last_run(token):
body = {
"page": 1,
"limit": 1,
"filter": [{"type": "equals", "field": "name", "value": "product.indexer"}],
}
data = api("POST", "/api/search/scheduled-task", token, body)
rows = data.get("data", [])
if not rows:
return None
last_run = rows[0].get("lastExecutionTime")
return parse_iso(last_run) if last_run else Noneasync function getProductIndexerLastRun(token) {
const body = {
page: 1,
limit: 1,
filter: [{ type: "equals", field: "name", value: "product.indexer" }],
};
const data = await api("POST", "/api/search/scheduled-task", token, body);
const rows = data.data || [];
if (rows.length === 0) return null;
const lastRun = rows[0].lastExecutionTime;
return lastRun ? new Date(lastRun) : null;
}Apply the action, only when told to write
When the decision is reindex, call POST /api/_action/index scoped to the affected products, which lets Shopware emit fresh EntityIndexingMessages without a full-catalog rebuild. When it is create_visibility or fix_visibility, write only the product-visibility row, never active or any state machine field. When it is flag_category_mismatch, only report it, since fixing category assignment is a merchandising decision, not something to automate blindly.
def trigger_reindex(token, product_ids):
api("POST", "/api/_action/index?entity=product", token, {"ids": product_ids} if product_ids else None)
def create_visibility(token, product_id, sales_channel_id, visibility=30):
body = {"productId": product_id, "salesChannelId": sales_channel_id, "visibility": visibility}
api("POST", "/api/product-visibility", token, body)
def fix_visibility(token, visibility_row_id, visibility=30):
api("PATCH", f"/api/product-visibility/{visibility_row_id}", token, {"visibility": visibility})async function triggerReindex(token, productIds) {
await api("POST", "/api/_action/index?entity=product", token, productIds.length ? { ids: productIds } : null);
}
async function createVisibility(token, productId, salesChannelId, visibility = 30) {
const body = { productId, salesChannelId, visibility };
await api("POST", "/api/product-visibility", token, body);
}
async function fixVisibility(token, visibilityRowId, visibility = 30) {
await api("PATCH", `/api/product-visibility/${visibilityRowId}`, token, { visibility });
}Always start with DRY_RUN=true. The script never PATCHes product.active or any state machine field, since those are not the fault here. It only creates or fixes a product-visibility row, or triggers a scoped reindex, and after triggering it, re-poll GET /api/search/scheduled-task for product.indexer until lastExecutionTime advances past the trigger time, rather than assuming the reindex finished the instant the request returned.
The full code
Here is the complete script in one file for each language. It reads settings from the environment, logs what it finds, respects the dry run flag, and is safe to run again and again because it only ever recommends a reindex, a visibility fix, or a category mismatch flag, and it never touches the fields that decide whether a product is truly active.
View this code on GitHub Full runnable folder with tests in the shopware-fixes repo.
"""Find Shopware 6 products that are active, assigned, and visible in the Admin API
yet absent from storefront listings and search, and repair them, safely.
Storefront listing and search read a denormalized, asynchronously built view of the
catalog, kept warm by product.indexer and, if enabled, Elasticsearch messages consumed
through messenger:consume. A product can be fully correct in the database, active,
assigned to a sales channel, with a product_visibility row at visibility 30, and linked
into the right category, yet still be absent from sales-channel.product.repository
results if the corresponding indexing message was queued but never consumed. This script
pulls active products with their visibilities and categories, reads the product.indexer
scheduled task's lastExecutionTime, and runs a pure decision function per product. In
dry run it only reports the recommended action. Only when told to write does it trigger
a scoped reindex, or, in the rarer case a visibility row is genuinely missing or wrong,
create or fix that one row. It never touches active or any state machine field.
Run on a schedule. Safe to run again and again.
"""
import os
import logging
import datetime
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("reconcile_absent_products")
SHOPWARE_URL = os.environ["SHOPWARE_URL"].rstrip("/")
CLIENT_ID = os.environ["SHOPWARE_CLIENT_ID"]
CLIENT_SECRET = os.environ["SHOPWARE_CLIENT_SECRET"]
STOREFRONT_SALES_CHANNEL_ID = os.environ.get("STOREFRONT_SALES_CHANNEL_ID", "")
REQUIRED_CATEGORY_IDS = [
c.strip() for c in os.environ.get("REQUIRED_CATEGORY_IDS", "").split(",") if c.strip()
]
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
def get_token():
r = requests.post(
f"{SHOPWARE_URL}/api/oauth/token",
json={
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
},
headers={"Accept": "application/json"},
timeout=30,
)
r.raise_for_status()
return r.json()["access_token"]
def api(method, path, token, json_body=None):
r = requests.request(
method,
f"{SHOPWARE_URL}{path}",
json=json_body,
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json",
"Content-Type": "application/json",
},
timeout=30,
)
r.raise_for_status()
return r.json() if r.text else {}
def parse_iso(value):
return datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
def search_active_products(token, page=1, limit=500):
body = {
"page": page,
"limit": limit,
"filter": [{"type": "equals", "field": "active", "value": True}],
"associations": {"visibilities": {}, "categories": {}},
"total-count-mode": 1,
}
return api("POST", "/api/search/product", token, body)
def get_product_indexer_last_run(token):
body = {
"page": 1,
"limit": 1,
"filter": [{"type": "equals", "field": "name", "value": "product.indexer"}],
}
data = api("POST", "/api/search/scheduled-task", token, body)
rows = data.get("data", [])
if not rows:
return None
last_run = rows[0].get("lastExecutionTime")
return parse_iso(last_run) if last_run else None
def decide_visibility_repair(product, target_sales_channel_id, required_category_ids,
indexer_last_run, last_write_at):
if not product.get("active", False):
return {"action": "none", "reason": "inactive, not a listing bug"}
row = next(
(v for v in (product.get("visibilities") or [])
if v.get("salesChannelId") == target_sales_channel_id),
None,
)
if row is None:
return {"action": "create_visibility", "reason": "no product_visibility row for sales channel"}
if row.get("visibility") != 30:
return {"action": "fix_visibility", "reason": "visibility below 30 (hidden from listing/search)"}
category_ids = set(product.get("categoryIds") or [])
if not set(required_category_ids).issubset(category_ids):
return {"action": "flag_category_mismatch", "reason": "not linked to storefront category/stream"}
if indexer_last_run < last_write_at:
return {"action": "reindex", "reason": "index stale relative to last DB write"}
return {"action": "none", "reason": "configuration already correct"}
def create_visibility(token, product_id, sales_channel_id, visibility=30):
body = {"productId": product_id, "salesChannelId": sales_channel_id, "visibility": visibility}
api("POST", "/api/product-visibility", token, body)
def fix_visibility(token, visibility_row_id, visibility=30):
api("PATCH", f"/api/product-visibility/{visibility_row_id}", token, {"visibility": visibility})
def trigger_reindex(token, product_ids):
api("POST", "/api/_action/index?entity=product", token, {"ids": product_ids} if product_ids else None)
def fetch_active_products(token):
products = []
page = 1
while True:
data = search_active_products(token, page=page)
batch = data.get("data", [])
for row in batch:
visibilities = [
{"id": v["id"], "salesChannelId": v["salesChannelId"], "visibility": v.get("visibility", 30)}
for v in (row.get("visibilities") or [])
]
category_ids = [c["id"] for c in (row.get("categories") or [])]
products.append({
"id": row["id"],
"productNumber": row.get("productNumber", ""),
"active": row.get("active", False),
"visibilities": visibilities,
"categoryIds": category_ids,
"updatedAt": row.get("updatedAt") or row.get("createdAt"),
})
if not batch or page * 500 >= data.get("total", 0):
break
page += 1
return products
def run():
if not STOREFRONT_SALES_CHANNEL_ID:
raise RuntimeError("Set STOREFRONT_SALES_CHANNEL_ID before running.")
token = get_token()
indexer_last_run = get_product_indexer_last_run(token)
if indexer_last_run is None:
raise RuntimeError("Could not read product.indexer scheduled task lastExecutionTime.")
products = fetch_active_products(token)
to_reindex = []
planned = 0
for product in products:
last_write_at = parse_iso(product["updatedAt"]) if product["updatedAt"] else indexer_last_run
decision = decide_visibility_repair(
product, STOREFRONT_SALES_CHANNEL_ID, REQUIRED_CATEGORY_IDS,
indexer_last_run, last_write_at,
)
if decision["action"] == "none":
continue
planned += 1
log.info(
"Product %s action=%s reason=%s. %s",
product["productNumber"] or product["id"], decision["action"], decision["reason"],
"would apply" if DRY_RUN else "applying",
)
if DRY_RUN:
continue
if decision["action"] == "reindex":
to_reindex.append(product["id"])
elif decision["action"] == "create_visibility":
create_visibility(token, product["id"], STOREFRONT_SALES_CHANNEL_ID)
elif decision["action"] == "fix_visibility":
row = next(
v for v in product["visibilities"] if v["salesChannelId"] == STOREFRONT_SALES_CHANNEL_ID
)
fix_visibility(token, row["id"])
# flag_category_mismatch is report-only, never auto-applied
if not DRY_RUN and to_reindex:
trigger_reindex(token, to_reindex)
log.info("Done. %d product(s) %s.", planned, "need action" if DRY_RUN else "processed")
if __name__ == "__main__":
run()/**
* Find Shopware 6 products that are active, assigned, and visible in the Admin API
* yet absent from storefront listings and search, and repair them, safely.
*
* Storefront listing and search read a denormalized, asynchronously built view of the
* catalog, kept warm by product.indexer and, if enabled, Elasticsearch messages consumed
* through messenger:consume. A product can be fully correct in the database, active,
* assigned to a sales channel, with a product_visibility row at visibility 30, and linked
* into the right category, yet still be absent from sales-channel.product.repository
* results if the corresponding indexing message was queued but never consumed. This script
* pulls active products with their visibilities and categories, reads the product.indexer
* scheduled task's lastExecutionTime, and runs a pure decision function per product. In
* dry run it only reports the recommended action. Only when told to write does it trigger
* a scoped reindex, or, in the rarer case a visibility row is genuinely missing or wrong,
* create or fix that one row. It never touches active or any state machine field.
* Run on a schedule. Safe to run again and again.
*/
import { pathToFileURL } from "node:url";
const SHOPWARE_URL = (process.env.SHOPWARE_URL || "https://demo.example.com").replace(/\/$/, "");
const CLIENT_ID = process.env.SHOPWARE_CLIENT_ID || "SWIAxxx";
const CLIENT_SECRET = process.env.SHOPWARE_CLIENT_SECRET || "secret_dummy";
const STOREFRONT_SALES_CHANNEL_ID = process.env.STOREFRONT_SALES_CHANNEL_ID || "";
const REQUIRED_CATEGORY_IDS = (process.env.REQUIRED_CATEGORY_IDS || "")
.split(",")
.map((c) => c.trim())
.filter(Boolean);
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
export function decideVisibilityRepair(product, targetSalesChannelId, requiredCategoryIds,
indexerLastRun, lastWriteAt) {
if (!product.active) {
return { action: "none", reason: "inactive, not a listing bug" };
}
const row = (product.visibilities || []).find(
(v) => v.salesChannelId === targetSalesChannelId
);
if (!row) {
return { action: "create_visibility", reason: "no product_visibility row for sales channel" };
}
if (row.visibility !== 30) {
return { action: "fix_visibility", reason: "visibility below 30 (hidden from listing/search)" };
}
const categoryIds = new Set(product.categoryIds || []);
const missingCategory = requiredCategoryIds.some((id) => !categoryIds.has(id));
if (missingCategory) {
return { action: "flag_category_mismatch", reason: "not linked to storefront category/stream" };
}
if (indexerLastRun < lastWriteAt) {
return { action: "reindex", reason: "index stale relative to last DB write" };
}
return { action: "none", reason: "configuration already correct" };
}
async function getToken() {
const res = await fetch(`${SHOPWARE_URL}/api/oauth/token`, {
method: "POST",
headers: { "Content-Type": "application/json", Accept: "application/json" },
body: JSON.stringify({
grant_type: "client_credentials",
client_id: CLIENT_ID,
client_secret: CLIENT_SECRET,
}),
});
if (!res.ok) throw new Error(`Shopware auth ${res.status}`);
const body = await res.json();
return body.access_token;
}
async function api(method, path, token, jsonBody) {
const res = await fetch(`${SHOPWARE_URL}${path}`, {
method,
headers: {
Authorization: `Bearer ${token}`,
Accept: "application/json",
"Content-Type": "application/json",
},
body: jsonBody ? JSON.stringify(jsonBody) : undefined,
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const text = await res.text();
return text ? JSON.parse(text) : {};
}
async function searchActiveProducts(token, page = 1, limit = 500) {
const body = {
page,
limit,
filter: [{ type: "equals", field: "active", value: true }],
associations: { visibilities: {}, categories: {} },
"total-count-mode": 1,
};
return api("POST", "/api/search/product", token, body);
}
async function getProductIndexerLastRun(token) {
const body = {
page: 1,
limit: 1,
filter: [{ type: "equals", field: "name", value: "product.indexer" }],
};
const data = await api("POST", "/api/search/scheduled-task", token, body);
const rows = data.data || [];
if (rows.length === 0) return null;
const lastRun = rows[0].lastExecutionTime;
return lastRun ? new Date(lastRun) : null;
}
async function createVisibility(token, productId, salesChannelId, visibility = 30) {
const body = { productId, salesChannelId, visibility };
await api("POST", "/api/product-visibility", token, body);
}
async function fixVisibility(token, visibilityRowId, visibility = 30) {
await api("PATCH", `/api/product-visibility/${visibilityRowId}`, token, { visibility });
}
async function triggerReindex(token, productIds) {
await api("POST", "/api/_action/index?entity=product", token, productIds.length ? { ids: productIds } : null);
}
async function fetchActiveProducts(token) {
const products = [];
let page = 1;
while (true) {
const data = await searchActiveProducts(token, page);
const batch = data.data || [];
for (const row of batch) {
const visibilities = (row.visibilities || []).map((v) => ({
id: v.id,
salesChannelId: v.salesChannelId,
visibility: v.visibility ?? 30,
}));
const categoryIds = (row.categories || []).map((c) => c.id);
products.push({
id: row.id,
productNumber: row.productNumber || "",
active: row.active || false,
visibilities,
categoryIds,
updatedAt: row.updatedAt || row.createdAt,
});
}
if (batch.length === 0 || page * 500 >= (data.total || 0)) break;
page++;
}
return products;
}
export async function run() {
if (!STOREFRONT_SALES_CHANNEL_ID) {
throw new Error("Set STOREFRONT_SALES_CHANNEL_ID before running.");
}
const token = await getToken();
const indexerLastRun = await getProductIndexerLastRun(token);
if (!indexerLastRun) {
throw new Error("Could not read product.indexer scheduled task lastExecutionTime.");
}
const products = await fetchActiveProducts(token);
const toReindex = [];
let planned = 0;
for (const product of products) {
const lastWriteAt = product.updatedAt ? new Date(product.updatedAt) : indexerLastRun;
const decision = decideVisibilityRepair(
product, STOREFRONT_SALES_CHANNEL_ID, REQUIRED_CATEGORY_IDS,
indexerLastRun, lastWriteAt
);
if (decision.action === "none") continue;
planned++;
console.log(
`Product ${product.productNumber || product.id} action=${decision.action} reason=${decision.reason}. ${DRY_RUN ? "would apply" : "applying"}`
);
if (DRY_RUN) continue;
if (decision.action === "reindex") {
toReindex.push(product.id);
} else if (decision.action === "create_visibility") {
await createVisibility(token, product.id, STOREFRONT_SALES_CHANNEL_ID);
} else if (decision.action === "fix_visibility") {
const row = product.visibilities.find((v) => v.salesChannelId === STOREFRONT_SALES_CHANNEL_ID);
await fixVisibility(token, row.id);
}
// flag_category_mismatch is report-only, never auto-applied
}
if (!DRY_RUN && toReindex.length) {
await triggerReindex(token, toReindex);
}
console.log(`Done. ${planned} product(s) ${DRY_RUN ? "need action" : "processed"}.`);
}
if (import.meta.url === pathToFileURL(process.argv[1]).href) {
run().catch((err) => { console.error(err); process.exit(1); });
}Add a test
The decision rule is the part most worth testing, because it decides whether a product gets reindexed, has its visibility repaired, or is left alone. Because decide_visibility_repair is pure and takes the clock as an input, the test needs no network and no Shopware instance. It just feeds in plain fixtures with fixed dates and checks the answer.
import datetime
from reconcile_absent_products import decide_visibility_repair
CHANNEL = "c" * 32
CATEGORY = "d" * 32
OLD = datetime.datetime(2026, 7, 1, tzinfo=datetime.timezone.utc)
NEW = datetime.datetime(2026, 7, 9, tzinfo=datetime.timezone.utc)
def product(**over):
base = {
"active": True,
"visibilities": [{"id": "v" * 32, "salesChannelId": CHANNEL, "visibility": 30}],
"categoryIds": [CATEGORY],
}
base.update(over)
return base
def test_none_when_inactive():
result = decide_visibility_repair(product(active=False), CHANNEL, [CATEGORY], NEW, OLD)
assert result["action"] == "none"
def test_create_visibility_when_row_missing():
result = decide_visibility_repair(product(visibilities=[]), CHANNEL, [CATEGORY], NEW, OLD)
assert result["action"] == "create_visibility"
def test_fix_visibility_when_below_30():
p = product(visibilities=[{"id": "v" * 32, "salesChannelId": CHANNEL, "visibility": 10}])
result = decide_visibility_repair(p, CHANNEL, [CATEGORY], NEW, OLD)
assert result["action"] == "fix_visibility"
def test_flag_category_mismatch_when_category_missing():
result = decide_visibility_repair(product(categoryIds=[]), CHANNEL, [CATEGORY], NEW, OLD)
assert result["action"] == "flag_category_mismatch"
def test_reindex_when_indexer_stale_relative_to_write():
result = decide_visibility_repair(product(), CHANNEL, [CATEGORY], OLD, NEW)
assert result["action"] == "reindex"
def test_none_when_everything_already_correct():
result = decide_visibility_repair(product(), CHANNEL, [CATEGORY], NEW, OLD)
assert result["action"] == "none"import { test } from "node:test";
import assert from "node:assert/strict";
import { decideVisibilityRepair } from "./reconcile-absent-products.js";
const CHANNEL = "c".repeat(32);
const CATEGORY = "d".repeat(32);
const OLD = new Date("2026-07-01T00:00:00Z");
const NEW = new Date("2026-07-09T00:00:00Z");
const product = (over = {}) => ({
active: true,
visibilities: [{ id: "v".repeat(32), salesChannelId: CHANNEL, visibility: 30 }],
categoryIds: [CATEGORY],
...over,
});
test("none when inactive", () => {
const result = decideVisibilityRepair(product({ active: false }), CHANNEL, [CATEGORY], NEW, OLD);
assert.equal(result.action, "none");
});
test("create_visibility when row missing", () => {
const result = decideVisibilityRepair(product({ visibilities: [] }), CHANNEL, [CATEGORY], NEW, OLD);
assert.equal(result.action, "create_visibility");
});
test("fix_visibility when below 30", () => {
const p = product({ visibilities: [{ id: "v".repeat(32), salesChannelId: CHANNEL, visibility: 10 }] });
const result = decideVisibilityRepair(p, CHANNEL, [CATEGORY], NEW, OLD);
assert.equal(result.action, "fix_visibility");
});
test("flag_category_mismatch when category missing", () => {
const result = decideVisibilityRepair(product({ categoryIds: [] }), CHANNEL, [CATEGORY], NEW, OLD);
assert.equal(result.action, "flag_category_mismatch");
});
test("reindex when indexer stale relative to write", () => {
const result = decideVisibilityRepair(product(), CHANNEL, [CATEGORY], OLD, NEW);
assert.equal(result.action, "reindex");
});
test("none when everything already correct", () => {
const result = decideVisibilityRepair(product(), CHANNEL, [CATEGORY], NEW, OLD);
assert.equal(result.action, "none");
});Case studies
A relaunch left messenger:consume off for two days
A fashion retailer redeployed their Shopware instance behind a new process manager. The web requests and the Admin came up fine, but the systemd unit running messenger:consume as a long-running worker was never migrated to the new host. Merchandisers kept publishing new season products, marking them active and assigning them to categories, and every check in the Admin looked perfect.
None of those new products showed up on the storefront. Running the reconciler in dry run found dozens of products whose product.indexer comparison came back stale, all created after the redeploy. Once the missing worker was restored and the scoped reindex ran for the affected ids, the entire backlog appeared on the storefront within minutes, with no data actually having been wrong the whole time.
A supplier feed import left visibility at 10 for a subset of SKUs
A hardware store ran a nightly supplier feed import that updated stock, price, and, occasionally, category assignment. A change in the supplier's category mapping caused the import tool to also touch product_visibility for a subset of SKUs, but a bug in that step wrote visibility: 10 (search only) instead of leaving the existing 30 in place. The products stayed searchable in the Admin, active and assigned, but vanished from category listing pages on the storefront.
The reconciler flagged these as fix_visibility rather than reindex, since the underlying row was present but wrong, not stale. The team fixed the import bug at the source and ran the reconciler once to correct the already-affected rows, restoring every SKU to its normal listing pages without anyone touching active or the category tree.
After this runs on a schedule, a product that is active, assigned, and visible in the Admin actually shows up where a merchandiser expects it. The script never guesses: it separates a genuinely wrong product_visibility row from a merely stale index, and it never touches active or a state machine field, since neither was ever the problem. The Admin keeps looking exactly the same, since it was already reading the correct table, but the storefront finally catches up with what the catalog owner already sees.
FAQ
Why does an active, assigned Shopware 6 product not show up in the storefront?
The storefront listing and search routes read a denormalized, asynchronously built view of the catalog, not the live product record. If the product.indexer or Elasticsearch message for that product was queued but never consumed, the read side never catches up with the write side, so the product looks correct in the Admin API yet is absent from sales-channel.product.repository results.
Is it safe to trigger a full reindex to fix this?
A full reindex works but is heavy on a large catalog. It is safer to scope the trigger to the affected entity, for example product.indexer, and confirm the scheduled task or message consumer actually drains the new messages afterward, rather than assuming the POST to the index action alone finished the job.
How do I know the message queue is the reason a product is missing, not a data problem?
Check that active is true, a product_visibility row exists with visibility 30 for the sales channel, and the category assignment is correct in the database. If all three are already right but the storefront still cannot find the product, compare the last write to the product against the lastExecutionTime of the product.indexer scheduled task. A stale indexer timestamp relative to a recent write means messenger:consume is not draining the queue.
Related field notes
Citations
On the problem:
- How does product visibility work in Shopware 6? matheusgontijo.com/2022/01/25/how-does-product-visibility-work-in-shopware-6
- Product visibility, Shopware Community Forum. forum.shopware.com/t/product-visibility/92398
- inactive products should be displayed, shopware/shopware issue #1219. github.com/shopware/shopware/issues/1219
On the solution:
- Shopware Admin API: the ProductVisibility entity. shopware.stoplight.io/docs/admin-api/8b3fa9c2e325c-product-visibility
- Shopware 6 Documentation: Configuration, Caches and Indexes. docs.shopware.com/en/shopware-6-en/configuration/caches-indexes
- Shopware changelog: add indexer skip support to Sync API and Indexer API call. github.com/shopware/shopware changelog release-6-4-5-0
Stuck on a tricky one?
If you have a problem in Shopware order states, stock, message queue, or data integrity that you would rather hand off, this is the kind of work I do. Message me and we can work through it together.
Did this bring a product back to life?
If this saved you a confusing missing-product ticket or a wasted afternoon in the Admin, you can buy me a coffee. It is the best way to keep these field notes free and growing.
Buy me a coffee on Ko-fi