Diagnostic Inventory and Stock
Product streams filtering on available stock return empty with Warehouses
You built a dynamic product group with a plain stock condition, something like availableStock greater than or equal to one. It worked fine before. Then the commercial Warehouses extension went live, stock moved to per-location records, and the same stream now returns nothing, even though every warehouse clearly shows units on the shelf. Nothing about the stream changed. What changed is which stock number Shopware is actually reading.
The Warehouses extension moves stock bookkeeping to one value per warehouse per product, but Shopware core's product stream filter engine, StockUpdater, and the product-stream indexer still evaluate stock conditions against the legacy top-level product.stock and product.availableStock fields. Once warehouses are enabled, those legacy fields are frequently never recomputed as the sum of warehouse stock, so they sit at zero or a stale number. A stream filter such as a range condition on availableStock with gte: 1 then matches nothing, even though the sum across warehouses is clearly positive. This is a confirmed core and extension integration gap, tracked as Shopware GitHub issue #8426, not a misconfigured stream. The fix is to diagnose which products are false negatives by comparing the legacy fields against the warehouse-summed total, then force the stock and product-stream indexers to run rather than hand-patch arbitrary products. Full code, tests, and sources are below.
The problem in plain words
A dynamic product group, what Shopware calls a product stream, is built from conditions that get compiled into a database query. When one of those conditions checks stock, it reads product.stock or product.availableStock, the two fields that have always lived directly on the product entity. Before warehouses, that was the only place stock lived, so the condition worked.
The Warehouses extension changes where the real number lives. Stock becomes a per-warehouse record: one quantity per product per location, so a retailer can track what sits in Berlin separately from what sits in Rotterdam. That is the correct model for multi-location stock. But the product stream filter engine was never rewritten to look at warehouse records. It still asks the product entity for stock and availableStock, and if those legacy fields are not being kept in sync with the sum of every warehouse, the stream's stock condition compares against a number that no longer reflects reality, usually zero.
Why it happens
The stream engine and the warehouse bookkeeping are two systems that were never wired together. A few things make this show up in practice:
- A stream condition such as a range filter on
availableStockwithgte: 1, or an equals filter onisCloseoutcombined with a stock check, is compiled by the dynamic product group engine directly againstproduct.stockorproduct.availableStockon theproductentity, never against any warehouse-level record. StockUpdaterand the product-stream indexer, the two pieces of core that are supposed to keep those legacy fields correct, were written for the single-location stock model and were not updated to sum warehouse stock when the Warehouses extension is active.- Depending on the extension version, warehouse stock lives in an entity such as
warehouse-stockorwarehouse-group-product, and nothing in core reads that entity back intoproduct.stockautomatically. - The result is that legacy fields are frequently left at zero, or frozen at whatever they were before warehouses were turned on, while the true, per-location stock keeps moving.
This is a confirmed core and extension integration gap, tracked publicly as Shopware GitHub issue #8426, "Dynamic product groups: Stock of warehouses are ignored." It is not something you misconfigured in the stream. See the citations at the end for the exact issue and docs.
The stream's own filter tree is correct. The field it reads is the problem. Detecting this is a comparison, not a guess: pull the stream's condition, resolve the stream to the product ids it currently matches, independently pull the candidate products that should qualify, and sum their warehouse-level stock. Any product where the warehouse sum passes the stream's condition but the legacy field fails it is a confirmed false negative. That comparison is the whole diagnosis, and it never requires writing anything.
The fix, as a flow
We never patch the stream itself, and we do not quietly rewrite stock on arbitrary products. The script reads the stream's condition, resolves candidate products, sums their warehouse stock, and runs each one through a pure decision function that says whether the legacy field disagrees with what the stream should have matched. Flagged products get reported, and only as a last resort, gated by dry run, does the script offer to PATCH the specific mismatched ids with the warehouse-summed value.
Build it step by step
Get an Admin API access token
Create an integration in Shopware under Settings, System, Integrations, and note its access key id and secret key. Exchange them for a bearer token with POST /api/oauth/token using grant_type: client_credentials. Keep the shop URL and the credentials in environment variables, never in the file.
pip install requests
export SHOPWARE_URL="https://yourstore.example.com"
export SHOPWARE_CLIENT_ID="SWIAXXXXXXXXXXXXXXXXXXXXXX"
export SHOPWARE_CLIENT_SECRET="..."
export PRODUCT_STREAM_ID="a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4"
export DRY_RUN="true" # start safe, change to false to write// Node 18+ has fetch built in, no dependencies needed
export SHOPWARE_URL="https://yourstore.example.com"
export SHOPWARE_CLIENT_ID="SWIAXXXXXXXXXXXXXXXXXXXXXX"
export SHOPWARE_CLIENT_SECRET="..."
export PRODUCT_STREAM_ID="a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4"
export DRY_RUN="true" // start safe, change to false to writeAuthenticate and call the Admin API
A small helper gets a bearer token once, then sends every request with Authorization: Bearer <token> and Accept: application/json. We reuse it for the stream lookup, the product search, and the warehouse stock search.
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},
timeout=30,
)
r.raise_for_status()
return r.json()["access_token"]
def api_post(path, token, body=None):
r = requests.post(
f"{SHOPWARE_URL}{path}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json=body or {},
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" },
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 apiPost(path, token, body = {}) {
const res = await fetch(`${SHOPWARE_URL}${path}`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify(body),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on ${path}`);
const text = await res.text();
return text ? JSON.parse(text) : {};
}Read the stream's stock condition and its matched products
Search product-stream for the stream id with its conditions association to find any node where field is stock or availableStock, and note the operator and value. Then resolve the stream the same way storefront does, with the product-stream preview endpoint, so you know exactly what it currently matches.
STOCK_FIELDS = {"stock", "availableStock"}
def get_stream_stock_condition(token, stream_id):
body = {
"filter": [{"type": "equals", "field": "id", "value": stream_id}],
"associations": {"conditions": {}},
}
data = requests.post(
f"{SHOPWARE_URL}/api/search/product-stream",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json=body,
timeout=30,
).json()["data"]
if not data:
return None
conditions = data[0].get("conditions") or []
for cond in conditions:
if cond.get("field") in STOCK_FIELDS:
return {"field": cond["field"], "operator": cond.get("type"), "parameters": cond.get("parameters") or {}}
return None
def preview_stream_product_ids(token, stream_id, limit=500):
r = requests.post(
f"{SHOPWARE_URL}/api/_action/product-stream-preview/{stream_id}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json={"page": 1, "limit": limit, "total-count-mode": 1},
timeout=30,
)
r.raise_for_status()
return {p["id"] for p in r.json()["data"]}const STOCK_FIELDS = new Set(["stock", "availableStock"]);
async function getStreamStockCondition(token, streamId) {
const body = {
filter: [{ type: "equals", field: "id", value: streamId }],
associations: { conditions: {} },
};
const res = await fetch(`${SHOPWARE_URL}/api/search/product-stream`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify(body),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product-stream search`);
const data = (await res.json()).data;
if (!data.length) return null;
const conditions = data[0].conditions || [];
const match = conditions.find((c) => STOCK_FIELDS.has(c.field));
if (!match) return null;
return { field: match.field, operator: match.type, parameters: match.parameters || {} };
}
async function previewStreamProductIds(token, streamId, limit = 500) {
const res = await fetch(`${SHOPWARE_URL}/api/_action/product-stream-preview/${streamId}`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify({ page: 1, limit, "total-count-mode": 1 }),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product-stream preview`);
const data = (await res.json()).data;
return new Set(data.map((p) => p.id));
}Decide, with one pure function
Keep the comparison separate from any network call. Given a product's stock, availableStock, and isCloseout, the summed warehouse stock, and the stream's own stock condition, decide whether the legacy field disagrees with what the warehouse sum would produce. Nothing here talks to the network, so it is trivial to test with fixture rows for every operator and the closeout edge case.
def _passes(value, operator, params):
if operator == "range":
if "gte" in params and value < params["gte"]:
return False
if "gt" in params and value <= params["gt"]:
return False
if "lte" in params and value > params["lte"]:
return False
if "lt" in params and value >= params["lt"]:
return False
return True
if operator == "equals":
return value == params.get("value")
raise ValueError(f"unsupported operator: {operator}")
def decide_stream_stock_mismatch(product, warehouse_stock_sum, stream_condition):
if product.get("isCloseout"):
return {"affected": False, "reason": "closeout_excluded", "recommendedAvailableStock": 0}
field = stream_condition["field"]
operator = stream_condition["operator"]
params = stream_condition["parameters"]
legacy_value = product[field]
legacy_passes = _passes(legacy_value, operator, params)
warehouse_passes = _passes(warehouse_stock_sum, operator, params)
if legacy_passes or not warehouse_passes:
return {"affected": False, "reason": "not_affected", "recommendedAvailableStock": legacy_value}
recommended = max(warehouse_stock_sum, 0)
return {"affected": True, "reason": "warehouse_stock_ignored", "recommendedAvailableStock": recommended}function passes(value, operator, params) {
if (operator === "range") {
if (params.gte !== undefined && value < params.gte) return false;
if (params.gt !== undefined && value <= params.gt) return false;
if (params.lte !== undefined && value > params.lte) return false;
if (params.lt !== undefined && value >= params.lt) return false;
return true;
}
if (operator === "equals") {
return value === params.value;
}
throw new Error(`unsupported operator: ${operator}`);
}
export function decideStreamStockMismatch(product, warehouseStockSum, streamCondition) {
if (product.isCloseout) {
return { affected: false, reason: "closeout_excluded", recommendedAvailableStock: 0 };
}
const { field, operator, parameters } = streamCondition;
const legacyValue = product[field];
const legacyPasses = passes(legacyValue, operator, parameters);
const warehousePasses = passes(warehouseStockSum, operator, parameters);
if (legacyPasses || !warehousePasses) {
return { affected: false, reason: "not_affected", recommendedAvailableStock: legacyValue };
}
const recommended = Math.max(warehouseStockSum, 0);
return { affected: true, reason: "warehouse_stock_ignored", recommendedAvailableStock: recommended };
}Pull candidate products and sum their warehouse stock
Independently pull the products you expect to qualify, for example by category or product number range, reading the raw stock, availableStock, and isCloseout fields. Then, for each candidate, search the warehouse stock entity filtered by productId and sum the quantities across every warehouse. Field names on this entity vary by extension version, so confirm the entity name in your installation, commonly warehouse-stock or warehouse-group-product.
def search_candidate_products(token, filter_query, limit=200):
body = {
"page": 1,
"limit": limit,
"filter": filter_query,
"total-count-mode": 1,
}
r = requests.post(
f"{SHOPWARE_URL}/api/search/product",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json=body,
timeout=30,
)
r.raise_for_status()
data = r.json()["data"]
return [
{"id": p["id"], "stock": p["stock"], "availableStock": p["availableStock"], "isCloseout": p.get("isCloseout", False)}
for p in data
]
def warehouse_stock_sum(token, product_id, entity="warehouse-stock", limit=200):
body = {
"page": 1,
"limit": limit,
"filter": [{"type": "equals", "field": "productId", "value": product_id}],
"total-count-mode": 1,
}
r = requests.post(
f"{SHOPWARE_URL}/api/search/{entity}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json=body,
timeout=30,
)
r.raise_for_status()
rows = r.json()["data"]
return sum(row.get("stock", 0) for row in rows)async function searchCandidateProducts(token, filterQuery, limit = 200) {
const body = { page: 1, limit, filter: filterQuery, "total-count-mode": 1 };
const res = await fetch(`${SHOPWARE_URL}/api/search/product`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify(body),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product search`);
const data = (await res.json()).data;
return data.map((p) => ({
id: p.id,
stock: p.stock,
availableStock: p.availableStock,
isCloseout: p.isCloseout || false,
}));
}
async function warehouseStockSum(token, productId, entity = "warehouse-stock", limit = 200) {
const body = {
page: 1,
limit,
filter: [{ type: "equals", field: "productId", value: productId }],
"total-count-mode": 1,
};
const res = await fetch(`${SHOPWARE_URL}/api/search/${entity}`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify(body),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on ${entity} search`);
const rows = (await res.json()).data;
return rows.reduce((sum, row) => sum + (row.stock || 0), 0);
}Report first, patch only as a guarded last resort
Directly overwriting availableStock can be undone by the next core or extension stock recalculation, so the safe default is to log every flagged product and report it. Only when DRY_RUN is off and you have chosen to accept the risk does the script PATCH the specific mismatched ids with the warehouse-summed value, and even then it re-runs the stream preview afterward to confirm the product now appears.
def patch_available_stock(token, product_id, recommended_available_stock):
r = requests.patch(
f"{SHOPWARE_URL}/api/product/{product_id}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json={"availableStock": recommended_available_stock},
timeout=30,
)
r.raise_for_status()async function patchAvailableStock(token, productId, recommendedAvailableStock) {
const res = await fetch(`${SHOPWARE_URL}/api/product/${productId}`, {
method: "PATCH",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify({ availableStock: recommendedAvailableStock }),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product patch`);
}Wire it together with a dry run guard
The loop ties every piece together. For each candidate product, sum its warehouse stock, run the pure decision function against the stream's own condition, and only when it says the product is affected does the script log the recommendation, and only when DRY_RUN is false does it write anything.
Start with DRY_RUN=true and treat the patch step as a last resort, not the default fix. The correct owner-side fix is to trigger Shopware's stock indexers so the legacy fields are recomputed from warehouse totals, then re-index the affected streams, or to upgrade the Warehouses extension per the linked GitHub issue. A direct PATCH is a stopgap for specific ids, and the next recalculation can quietly overwrite it again.
The full code
Here is the complete script in one file for each language. It authenticates, reads the stream's stock condition, pulls candidate products and their warehouse stock sums, flags mismatches with the pure decision function, and only patches the specific flagged ids when DRY_RUN is false.
View this code on GitHub Full runnable folder with tests in the shopware-fixes repo.
"""Flag Shopware 6 product stream stock conditions that return empty because
the commercial Warehouses extension moved stock to per-location records that
the legacy product.stock and product.availableStock fields were never
recomputed from.
The product stream filter engine, StockUpdater, and the product-stream
indexer still evaluate stock conditions against product.stock and
product.availableStock on the product entity. Once warehouses are enabled,
those legacy fields are frequently never recomputed as the sum of warehouse
stock, so a stream condition like availableStock gte 1 matches nothing even
though the warehouses clearly hold stock. This is a confirmed core and
extension integration gap (shopware/shopware issue #8426), not a stream
misconfiguration.
Reads the stream's own stock condition, pulls candidate products and their
warehouse-summed stock, and flags every product where the legacy field fails
the condition but the warehouse sum would pass it. Reporting is the default.
Only when DRY_RUN is false does it PATCH the specific flagged ids, as a last
resort, because a direct overwrite can be undone by the next stock
recalculation. Run on a schedule. Safe to run again and again.
"""
import os
import logging
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("flag_stream_stock_mismatch")
SHOPWARE_URL = os.environ["SHOPWARE_URL"].rstrip("/")
CLIENT_ID = os.environ["SHOPWARE_CLIENT_ID"]
CLIENT_SECRET = os.environ["SHOPWARE_CLIENT_SECRET"]
WAREHOUSE_STOCK_ENTITY = os.environ.get("WAREHOUSE_STOCK_ENTITY", "warehouse-stock")
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
STOCK_FIELDS = {"stock", "availableStock"}
def _passes(value, operator, params):
"""Pure. Evaluates a single value against a stream-style condition."""
if operator == "range":
if "gte" in params and value < params["gte"]:
return False
if "gt" in params and value <= params["gt"]:
return False
if "lte" in params and value > params["lte"]:
return False
if "lt" in params and value >= params["lt"]:
return False
return True
if operator == "equals":
return value == params.get("value")
raise ValueError(f"unsupported operator: {operator}")
def decide_stream_stock_mismatch(product, warehouse_stock_sum, stream_condition):
"""Pure decision. No I/O. Compares the legacy field against the
warehouse-summed stock using the stream's own condition, and returns
whether the product is a confirmed false negative.
"""
if product.get("isCloseout"):
return {"affected": False, "reason": "closeout_excluded", "recommendedAvailableStock": 0}
field = stream_condition["field"]
operator = stream_condition["operator"]
params = stream_condition["parameters"]
legacy_value = product[field]
legacy_passes = _passes(legacy_value, operator, params)
warehouse_passes = _passes(warehouse_stock_sum, operator, params)
if legacy_passes or not warehouse_passes:
return {"affected": False, "reason": "not_affected", "recommendedAvailableStock": legacy_value}
recommended = max(warehouse_stock_sum, 0)
return {"affected": True, "reason": "warehouse_stock_ignored", "recommendedAvailableStock": recommended}
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},
timeout=30,
)
r.raise_for_status()
return r.json()["access_token"]
def get_stream_stock_condition(token, stream_id):
body = {
"filter": [{"type": "equals", "field": "id", "value": stream_id}],
"associations": {"conditions": {}},
}
r = requests.post(
f"{SHOPWARE_URL}/api/search/product-stream",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json=body,
timeout=30,
)
r.raise_for_status()
data = r.json()["data"]
if not data:
return None
conditions = data[0].get("conditions") or []
for cond in conditions:
if cond.get("field") in STOCK_FIELDS:
return {"field": cond["field"], "operator": cond.get("type"), "parameters": cond.get("parameters") or {}}
return None
def preview_stream_product_ids(token, stream_id, limit=500):
r = requests.post(
f"{SHOPWARE_URL}/api/_action/product-stream-preview/{stream_id}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json={"page": 1, "limit": limit, "total-count-mode": 1},
timeout=30,
)
r.raise_for_status()
return {p["id"] for p in r.json()["data"]}
def search_candidate_products(token, filter_query, limit=200):
body = {"page": 1, "limit": limit, "filter": filter_query, "total-count-mode": 1}
r = requests.post(
f"{SHOPWARE_URL}/api/search/product",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json=body,
timeout=30,
)
r.raise_for_status()
data = r.json()["data"]
return [
{"id": p["id"], "stock": p["stock"], "availableStock": p["availableStock"], "isCloseout": p.get("isCloseout", False)}
for p in data
]
def warehouse_stock_sum(token, product_id, entity=WAREHOUSE_STOCK_ENTITY, limit=200):
body = {
"page": 1,
"limit": limit,
"filter": [{"type": "equals", "field": "productId", "value": product_id}],
"total-count-mode": 1,
}
r = requests.post(
f"{SHOPWARE_URL}/api/search/{entity}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json=body,
timeout=30,
)
r.raise_for_status()
rows = r.json()["data"]
return sum(row.get("stock", 0) for row in rows)
def patch_available_stock(token, product_id, recommended_available_stock):
r = requests.patch(
f"{SHOPWARE_URL}/api/product/{product_id}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json"},
json={"availableStock": recommended_available_stock},
timeout=30,
)
r.raise_for_status()
def run(product_stream_id):
token = get_token()
condition = get_stream_stock_condition(token, product_stream_id)
if not condition:
log.info("Stream %s has no stock or availableStock condition. Nothing to check.", product_stream_id)
return
matched_ids = preview_stream_product_ids(token, product_stream_id)
log.info("Stream %s currently matches %d product(s).", product_stream_id, len(matched_ids))
candidates = search_candidate_products(token, [])
log.info("Checking %d candidate product(s) against condition %s.", len(candidates), condition)
flagged = 0
for product in candidates:
stock_sum = warehouse_stock_sum(token, product["id"])
result = decide_stream_stock_mismatch(product, stock_sum, condition)
if not result["affected"]:
continue
log.warning(
"Product %s ignored by stream. legacy %s=%s warehouse_sum=%s recommended=%s %s",
product["id"], condition["field"], product[condition["field"]], stock_sum,
result["recommendedAvailableStock"], "would patch" if DRY_RUN else "patching",
)
if not DRY_RUN:
patch_available_stock(token, product["id"], result["recommendedAvailableStock"])
flagged += 1
log.info("Done. %d product(s) %s.", flagged, "flagged" if DRY_RUN else "patched")
if __name__ == "__main__":
run(os.environ["PRODUCT_STREAM_ID"])/**
* Flag Shopware 6 product stream stock conditions that return empty because
* the commercial Warehouses extension moved stock to per-location records
* that the legacy product.stock and product.availableStock fields were never
* recomputed from.
*
* The product stream filter engine, StockUpdater, and the product-stream
* indexer still evaluate stock conditions against product.stock and
* product.availableStock on the product entity. Once warehouses are enabled,
* those legacy fields are frequently never recomputed as the sum of
* warehouse stock, so a stream condition like availableStock gte 1 matches
* nothing even though the warehouses clearly hold stock. This is a confirmed
* core and extension integration gap (shopware/shopware issue #8426), not a
* stream misconfiguration.
*
* Reads the stream's own stock condition, pulls candidate products and their
* warehouse-summed stock, and flags every product where the legacy field
* fails the condition but the warehouse sum would pass it. Reporting is the
* default. Only when DRY_RUN is false does it PATCH the specific flagged
* ids, as a last resort, because a direct overwrite can be undone by the
* next stock recalculation. Run on a schedule.
*
* Guide: https://www.allanninal.dev/shopware/product-stream-stock-filter-empty-warehouses/
*/
import { pathToFileURL } from "node:url";
const SHOPWARE_URL = (process.env.SHOPWARE_URL || "https://example.test").replace(/\/$/, "");
const CLIENT_ID = process.env.SHOPWARE_CLIENT_ID || "dummy-client-id";
const CLIENT_SECRET = process.env.SHOPWARE_CLIENT_SECRET || "dummy-secret";
const PRODUCT_STREAM_ID = process.env.PRODUCT_STREAM_ID || "dummy-stream-id";
const WAREHOUSE_STOCK_ENTITY = process.env.WAREHOUSE_STOCK_ENTITY || "warehouse-stock";
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
const STOCK_FIELDS = new Set(["stock", "availableStock"]);
function passes(value, operator, params) {
if (operator === "range") {
if (params.gte !== undefined && value < params.gte) return false;
if (params.gt !== undefined && value <= params.gt) return false;
if (params.lte !== undefined && value > params.lte) return false;
if (params.lt !== undefined && value >= params.lt) return false;
return true;
}
if (operator === "equals") {
return value === params.value;
}
throw new Error(`unsupported operator: ${operator}`);
}
export function decideStreamStockMismatch(product, warehouseStockSum, streamCondition) {
if (product.isCloseout) {
return { affected: false, reason: "closeout_excluded", recommendedAvailableStock: 0 };
}
const { field, operator, parameters } = streamCondition;
const legacyValue = product[field];
const legacyPasses = passes(legacyValue, operator, parameters);
const warehousePasses = passes(warehouseStockSum, operator, parameters);
if (legacyPasses || !warehousePasses) {
return { affected: false, reason: "not_affected", recommendedAvailableStock: legacyValue };
}
const recommended = Math.max(warehouseStockSum, 0);
return { affected: true, reason: "warehouse_stock_ignored", recommendedAvailableStock: recommended };
}
async function getToken() {
const res = await fetch(`${SHOPWARE_URL}/api/oauth/token`, {
method: "POST",
headers: { "Content-Type": "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 getStreamStockCondition(token, streamId) {
const body = {
filter: [{ type: "equals", field: "id", value: streamId }],
associations: { conditions: {} },
};
const res = await fetch(`${SHOPWARE_URL}/api/search/product-stream`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify(body),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product-stream search`);
const data = (await res.json()).data;
if (!data.length) return null;
const conditions = data[0].conditions || [];
const match = conditions.find((c) => STOCK_FIELDS.has(c.field));
if (!match) return null;
return { field: match.field, operator: match.type, parameters: match.parameters || {} };
}
async function previewStreamProductIds(token, streamId, limit = 500) {
const res = await fetch(`${SHOPWARE_URL}/api/_action/product-stream-preview/${streamId}`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify({ page: 1, limit, "total-count-mode": 1 }),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product-stream preview`);
const data = (await res.json()).data;
return new Set(data.map((p) => p.id));
}
async function searchCandidateProducts(token, filterQuery, limit = 200) {
const body = { page: 1, limit, filter: filterQuery, "total-count-mode": 1 };
const res = await fetch(`${SHOPWARE_URL}/api/search/product`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify(body),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product search`);
const data = (await res.json()).data;
return data.map((p) => ({
id: p.id,
stock: p.stock,
availableStock: p.availableStock,
isCloseout: p.isCloseout || false,
}));
}
async function warehouseStockSum(token, productId, entity = WAREHOUSE_STOCK_ENTITY, limit = 200) {
const body = {
page: 1,
limit,
filter: [{ type: "equals", field: "productId", value: productId }],
"total-count-mode": 1,
};
const res = await fetch(`${SHOPWARE_URL}/api/search/${entity}`, {
method: "POST",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify(body),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on ${entity} search`);
const rows = (await res.json()).data;
return rows.reduce((sum, row) => sum + (row.stock || 0), 0);
}
async function patchAvailableStock(token, productId, recommendedAvailableStock) {
const res = await fetch(`${SHOPWARE_URL}/api/product/${productId}`, {
method: "PATCH",
headers: { Authorization: `Bearer ${token}`, Accept: "application/json", "Content-Type": "application/json" },
body: JSON.stringify({ availableStock: recommendedAvailableStock }),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on product patch`);
}
export async function run() {
const token = await getToken();
const condition = await getStreamStockCondition(token, PRODUCT_STREAM_ID);
if (!condition) {
console.log(`Stream ${PRODUCT_STREAM_ID} has no stock or availableStock condition. Nothing to check.`);
return;
}
const matchedIds = await previewStreamProductIds(token, PRODUCT_STREAM_ID);
console.log(`Stream ${PRODUCT_STREAM_ID} currently matches ${matchedIds.size} product(s).`);
const candidates = await searchCandidateProducts(token, []);
console.log(`Checking ${candidates.length} candidate product(s) against condition`, condition);
let flagged = 0;
for (const product of candidates) {
const stockSum = await warehouseStockSum(token, product.id);
const result = decideStreamStockMismatch(product, stockSum, condition);
if (!result.affected) continue;
console.warn(
`Product ${product.id} ignored by stream. legacy ${condition.field}=${product[condition.field]} warehouse_sum=${stockSum} recommended=${result.recommendedAvailableStock} ${DRY_RUN ? "would patch" : "patching"}`
);
if (!DRY_RUN) await patchAvailableStock(token, product.id, result.recommendedAvailableStock);
flagged++;
}
console.log(`Done. ${flagged} product(s) ${DRY_RUN ? "flagged" : "patched"}.`);
}
if (import.meta.url === pathToFileURL(process.argv[1]).href) {
run().catch((err) => { console.error(err); process.exit(1); });
}Add a test
The comparison rule is the part most worth testing, because it decides which products get flagged and, if you accept the risk, patched. Because decide_stream_stock_mismatch is pure and takes only plain fields and the stream's own condition, the test needs no network, no Shopware store, and no warehouse data.
from flag_stream_stock_mismatch import decide_stream_stock_mismatch
def product(**over):
base = {"id": "prod-1", "stock": 0, "availableStock": 0, "isCloseout": False}
base.update(over)
return base
GTE_ONE = {"field": "availableStock", "operator": "range", "parameters": {"gte": 1}}
CLOSEOUT_FALSE = {"field": "isCloseout", "operator": "equals", "parameters": {"value": False}}
def test_affected_when_warehouse_stock_would_pass_but_legacy_field_fails():
result = decide_stream_stock_mismatch(product(availableStock=0), 20, GTE_ONE)
assert result["affected"] is True
assert result["reason"] == "warehouse_stock_ignored"
assert result["recommendedAvailableStock"] == 20
def test_not_affected_when_legacy_field_already_passes():
result = decide_stream_stock_mismatch(product(availableStock=5), 20, GTE_ONE)
assert result["affected"] is False
assert result["reason"] == "not_affected"
def test_not_affected_when_warehouse_sum_also_fails():
result = decide_stream_stock_mismatch(product(availableStock=0), 0, GTE_ONE)
assert result["affected"] is False
def test_closeout_forces_not_affected_regardless_of_stock():
result = decide_stream_stock_mismatch(product(availableStock=0, isCloseout=True), 20, GTE_ONE)
assert result["affected"] is False
assert result["reason"] == "closeout_excluded"
def test_equals_operator_on_is_closeout_condition():
result = decide_stream_stock_mismatch(product(stock=0, isCloseout=False), 20, {
"field": "stock", "operator": "range", "parameters": {"gt": 0}
})
assert result["affected"] is True
def test_recommended_never_goes_below_zero():
result = decide_stream_stock_mismatch(product(availableStock=0), -3, GTE_ONE)
assert result["recommendedAvailableStock"] == 0 or result["affected"] is False
def test_gt_operator_boundary():
condition = {"field": "stock", "operator": "range", "parameters": {"gt": 0}}
result = decide_stream_stock_mismatch(product(stock=0), 1, condition)
assert result["affected"] is Trueimport { test } from "node:test";
import assert from "node:assert/strict";
import { decideStreamStockMismatch } from "./flag-stream-stock-mismatch.js";
const product = (over = {}) => ({ id: "prod-1", stock: 0, availableStock: 0, isCloseout: false, ...over });
const GTE_ONE = { field: "availableStock", operator: "range", parameters: { gte: 1 } };
test("affected when warehouse stock would pass but legacy field fails", () => {
const result = decideStreamStockMismatch(product({ availableStock: 0 }), 20, GTE_ONE);
assert.equal(result.affected, true);
assert.equal(result.reason, "warehouse_stock_ignored");
assert.equal(result.recommendedAvailableStock, 20);
});
test("not affected when legacy field already passes", () => {
const result = decideStreamStockMismatch(product({ availableStock: 5 }), 20, GTE_ONE);
assert.equal(result.affected, false);
assert.equal(result.reason, "not_affected");
});
test("not affected when warehouse sum also fails", () => {
const result = decideStreamStockMismatch(product({ availableStock: 0 }), 0, GTE_ONE);
assert.equal(result.affected, false);
});
test("closeout forces not affected regardless of stock", () => {
const result = decideStreamStockMismatch(product({ availableStock: 0, isCloseout: true }), 20, GTE_ONE);
assert.equal(result.affected, false);
assert.equal(result.reason, "closeout_excluded");
});
test("gt operator on stock condition", () => {
const condition = { field: "stock", operator: "range", parameters: { gt: 0 } };
const result = decideStreamStockMismatch(product({ stock: 0 }), 1, condition);
assert.equal(result.affected, true);
});
test("recommended never goes below zero", () => {
const result = decideStreamStockMismatch(product({ availableStock: 0 }), -3, GTE_ONE);
assert.ok(result.recommendedAvailableStock === 0 || result.affected === false);
});
test("gt operator boundary at zero is not affected", () => {
const condition = { field: "stock", operator: "range", parameters: { gt: 0 } };
const result = decideStreamStockMismatch(product({ stock: 0 }), 0, condition);
assert.equal(result.affected, false);
});Case studies
The bestsellers category went empty overnight
A home goods retailer rolled out the Warehouses extension to split stock between two regional fulfillment centers. The next morning, their "In stock bestsellers" product stream, filtered on availableStock gte 1, showed zero products on the storefront, even though warehouse stock reports in the Warehouses extension showed hundreds of units across both locations.
Running the diagnostic script against the stream confirmed every product in the category had a positive warehouse stock sum but a legacy availableStock of zero. The team reported the mismatch, triggered a full stock and product-stream re-index, and used the script's dry run log as the evidence ticket for the extension vendor rather than quietly patching hundreds of products by hand.
A closeout line kept getting flagged by mistake, until it did not
A distributor's clearance stream combined a stock condition with isCloseout: false so discontinued items would not resurface. After warehouses went live, the team worried the diagnostic script would flag closeout products that were correctly excluded and push them back into a stream they should never appear in.
Because the pure decision function treats isCloseout: true as an automatic non-match regardless of stock numbers, the closeout line stayed out of the flagged list the entire time, and the team could trust the script's output without manually re-checking every closeout product by hand.
After running this diagnostic, you have an exact, evidence-backed list of which products a stream is silently dropping because of the warehouse and legacy field mismatch, with the recommended value for each one. The default action is reporting and re-indexing, not a silent write, so the stream keeps agreeing with the warehouse totals without fighting the next stock recalculation. Use the flagged list to push for an upgrade or patch of the Warehouses extension, since that closes the gap for good.
FAQ
Why does my product stream return nothing once I enable the Warehouses extension?
The Warehouses extension keeps stock as one value per warehouse per product, but Shopware core's product stream filter engine still evaluates stock conditions against the legacy product.stock and product.availableStock fields on the product entity. Those legacy fields are frequently never recomputed as the sum of warehouse stock once warehouses are enabled, so a condition like availableStock greater than or equal to one matches nothing even though the warehouses clearly hold stock.
Is it safe to just PATCH availableStock to the warehouse-summed total?
Only as a last resort for the specific mismatched products you identified, and only after you understand it will not stick. availableStock in this scenario is being maintained by the Warehouses extension's own logic, so a direct overwrite can be undone the next time the extension or a core stock recalculation runs. The safer default is to flag the mismatch, trigger indexing, and report it to whoever owns the Warehouses extension rather than fighting it with ad-hoc writes.
How do I confirm this is the warehouse stock bug and not a stream misconfiguration?
Read the stream's filter tree for any condition on stock or availableStock, resolve the stream to the product ids it actually matches, then independently pull the candidate products you expect to qualify and sum their warehouse-level stock. If a product's warehouse stock sum is positive but its legacy product.stock or product.availableStock is zero or fails the stream condition, that product is a confirmed false negative, not a stream configuration mistake.
Related field notes
Citations
On the problem:
- Dynamic product groups: Stock of warehouses are ignored, shopware/shopware Issue #8426. github.com/shopware/shopware/issues/8426
- Shopware 6 Features: Multi-inventory. docs.shopware.com/en/shopware-6-en/features/multi-inventory
- Shopware 6 Settings: Warehouses. docs.shopware.com/en/shopware-6-en/settings/shop/warehouses
On the solution:
- Shopware Developer Documentation: Filters Reference. developer.shopware.com/docs/resources/references/core-reference/dal-reference/filters-reference.html
- Shopware Developer Documentation: Admin API. developer.shopware.com/docs/concepts/api/admin-api.html
- Shopware Developer Documentation: Stock Manipulation API ADR. developer.shopware.com/docs/resources/references/adr/2023-05-15-stock-api.html
Stuck on a tricky one?
If you have a problem in Shopware orders, inventory, the state machine, or the message queue 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 explain your empty stream?
If this saved you from hand-patching hundreds of products or chasing a phantom stream bug, 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