feat(distro): RareWaves order ingest → cost-tracked NEW stock + approval landing

End-to-end: PRICEGOD scrapes a RareWaves order (account.rarewaves.com/orders/<id>) → POST /admin/intake/distro
→ resolve barcode→release_id (normalised: DB barcodes stored inconsistently, functional index on digits-only +
EAN/UPC leading-zero variants; Discogs barcode-search fallback) → stage one NEW copy per qty with cost_price +
cost_source, in_stock=false. Idempotent per order. New '/admin/newstock' approval queue: staff confirm/re-pick the
release (trgm search), set retail, flip new/used → publish (staged=f, in_stock=t). Bulk publish-all-matched-&-priced.
Backend verified e2e (ingest→queue→approve→live, unapproved stay staged). PRICEGOD: rarewaves.js content script +
account.rarewaves.com host perm + wowplatter.distro + bg rwDistro handler (needs extension reload to test live).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/admin_routes.py

@@ -190,6 +190,67 @@ async def timecards(days: int = Query(14), staff_id: int | None = None,
     return {"shifts": shifts, "totals": totals, "days": days}
 
 
+# ── New stock (distro purchases awaiting approval) — confirm release_id + set retail, then publish ──
+@router.get("/newstock")
+async def newstock_list(ident=Depends(require_token), db=Depends(get_db)):
+    rows = [dict(r) for r in (await db.execute(text("""
+        SELECT i.sku, i.title, i.identifier AS barcode, i.release_id, i.cost_price, i.price,
+               i.condition_type, i.cost_source, i.est_market_value,
+               COALESCE(dc.thumb, dr.thumb) AS thumb, dr.artists_sort AS artist, dr.year,
+               (i.attributes->>'resolved')::bool AS resolved, i.attributes->>'slug' AS slug
+        FROM inventory i
+        LEFT JOIN disc_cache dc ON dc.release_id = i.release_id
+        LEFT JOIN disc_release dr ON dr.id = i.release_id
+        WHERE i.store_id = :s AND i.staged AND i.cost_source IS NOT NULL
+        ORDER BY i.cost_source DESC, i.title NULLS LAST, i.sku"""),
+        {"s": ident["store_id"]})).mappings()]
+    return {"items": rows, "count": len(rows)}
+
+
+class NewStockIn(BaseModel):
+    release_id: int | None = None
+    title: str | None = None
+    price: float | None = None
+    condition_type: str | None = None
+    publish: bool = False
+
+
+@router.post("/newstock/{sku}")
+async def newstock_update(sku: str, body: NewStockIn, ident=Depends(require_token), db=Depends(get_db)):
+    """Staff approve a distro copy: re-pick the release / set retail / flip new-used, then publish (un-stage)."""
+    sets, params = [], {"sku": sku, "sid": ident["store_id"]}
+    if body.release_id is not None:
+        sets.append("release_id = :rid"); params["rid"] = body.release_id
+    if body.title is not None:
+        sets.append("title = :t"); params["t"] = body.title
+    if body.price is not None:
+        sets.append("price = :p"); params["p"] = body.price
+    if body.condition_type in ("new", "used"):
+        sets.append("condition_type = :ct"); params["ct"] = body.condition_type
+    if body.publish:
+        sets += ["staged = false", "status = 'publish'", "in_stock = true"]
+    if not sets:
+        return {"ok": True, "unchanged": True}
+    sets.append("updated_at = now()")
+    r = await db.execute(text(
+        f"UPDATE inventory SET {', '.join(sets)} WHERE sku = :sku AND store_id = :sid"), params)
+    await db.commit()
+    if not r.rowcount:
+        raise HTTPException(404, "not found")
+    return {"ok": True, "published": body.publish}
+
+
+@router.post("/newstock/publish-all")
+async def newstock_publish_all(ident=Depends(require_token), db=Depends(get_db)):
+    """Publish every resolved copy that already has a retail price (the bulk 'ship it' button)."""
+    r = await db.execute(text("""
+        UPDATE inventory SET staged = false, status = 'publish', in_stock = true, updated_at = now()
+        WHERE store_id = :s AND staged AND cost_source IS NOT NULL
+          AND release_id IS NOT NULL AND price IS NOT NULL"""), {"s": ident["store_id"]})
+    await db.commit()
+    return {"ok": True, "published": r.rowcount}
+
+
 @router.get("/inventory")
 async def inventory(q: str = Query(""), kind: str = Query(""), crate_id: int | None = Query(None),
                     page: int = Query(1, ge=1), ident=Depends(require_token), db=Depends(get_db)):

app/intake_routes.py

@@ -611,14 +611,19 @@ ITEM_IMG_DIR = Path(os.getenv("DISC_IMAGE_DIR", "/app/disc_images")) / "items"
 
 
 async def _resolve_barcode(db, barcode):
-    """barcode → release_id. Local disc_release_identifier first (16.5k 'Barcode' rows, instant),
-    then the Discogs barcode search on a miss."""
+    """barcode → release_id. Local disc_release_identifier first — NORMALISED (DB barcodes are stored
+    inconsistently: '5 018775 901762' vs '042285768916'), matched on a functional index over the
+    digits-only form, trying the EAN-13/UPC-A leading-zero variants. Discogs barcode search on a miss."""
     if not barcode:
         return None
-    bc = barcode.strip()
+    digits = re.sub(r"\D", "", str(barcode))
+    if len(digits) < 6:
+        return None
+    cands = list({digits, digits.lstrip("0"), "0" + digits})
     row = (await db.execute(text(
-        "SELECT release_id FROM disc_release_identifier WHERE value=:b AND type='Barcode' LIMIT 1"),
-        {"b": bc})).first()
+        "SELECT release_id FROM disc_release_identifier "
+        "WHERE type='Barcode' AND regexp_replace(value,'[^0-9]','','g') = ANY(:c) LIMIT 1"),
+        {"c": cands})).first()
     if row:
         return row[0]
     try:
@@ -626,7 +631,7 @@ async def _resolve_barcode(db, barcode):
         async with c:
             if not ok:
                 return None
-            r = await c.get("/database/search", params={"barcode": bc, "type": "release", "per_page": 1})
+            r = await c.get("/database/search", params={"barcode": digits, "type": "release", "per_page": 1})
             if r.status_code == 200:
                 res = r.json().get("results", [])
                 if res:
@@ -714,6 +719,72 @@ async def intake_scan(body: ScanIn, ident=Depends(require_token), db=Depends(get
     return {"staged": staged, "errors": errors}
 
 
+# --- Distro purchase ingest: scrape a distributor order → cost-tracked NEW stock ------------------
+# RareWaves (Shopify) order pages give barcode (in the /products/<ean>- handle) + title + qty +
+# unit cost ($X.XX/ea). The PRICEGOD extension scrapes the order and POSTs it here; we resolve the
+# barcode → release_id, stage one NEW copy per qty with cost_price + cost_source, for staff approval.
+
+class DistroItem(BaseModel):
+    barcode: str | None = None
+    release_id: int | None = None
+    title: str | None = None
+    artist: str | None = None
+    year: int | None = None
+    qty: int = 1
+    unit_cost: float | None = None
+    slug: str | None = None
+    variant_id: str | None = None
+    image: str | None = None
+
+
+class DistroIn(BaseModel):
+    source: str = "rarewaves"
+    order_ref: str
+    items: list[DistroItem]
+
+
+@router.post("/distro")
+async def intake_distro(body: DistroIn, ident=Depends(require_token), db=Depends(get_db)):
+    """Ingest a distributor order → staged NEW stock with per-copy cost. Idempotent per order
+    (re-scraping the same order is a no-op). Unresolved barcodes still stage (staff pick the release)."""
+    sid = ident["store_id"]
+    cost_source = f"{body.source} #{body.order_ref}"
+    existing = (await db.execute(text(
+        "SELECT count(*) FROM inventory WHERE cost_source=:cs AND store_id=:sid"),
+        {"cs": cost_source, "sid": sid})).scalar()
+    if existing:
+        return {"ok": True, "already_ingested": True, "existing": existing, "cost_source": cost_source}
+
+    staged, errors = [], []
+    for it in body.items:
+        try:
+            rid = it.release_id or await _resolve_barcode(db, it.barcode)
+            meta = await _enrich(db, rid) if rid else None
+            title = it.title or (meta or {}).get("title")
+            weight = (meta or {}).get("weight")
+            for _ in range(max(1, it.qty)):
+                sku = _new_sku()
+                attrs = {"source": body.source, "order_ref": body.order_ref, "slug": it.slug,
+                         "variant_id": it.variant_id, "year": it.year,
+                         "resolved": rid is not None, "scrape_image": it.image}
+                await db.execute(text("""
+                    INSERT INTO inventory (sku, store_id, kind, release_id, identifier, title,
+                        cost_price, cost_source, condition_type, condition, weight_g, attributes,
+                        staged, in_stock, status)
+                    VALUES (:sku,:sid,'vinyl',:rid,:bc,:title,:cost,:cs,'new','M',:wt,
+                            CAST(:attrs AS jsonb), true, false, 'staged')
+                    ON CONFLICT (sku) DO NOTHING"""),
+                    {"sku": sku, "sid": sid, "rid": rid, "bc": it.barcode, "title": title,
+                     "cost": it.unit_cost, "cs": cost_source, "wt": weight, "attrs": json.dumps(attrs)})
+                staged.append({"sku": sku, "release_id": rid, "title": title, "resolved": rid is not None})
+        except Exception as e:
+            errors.append({"barcode": it.barcode, "error": str(e)})
+    await db.commit()
+    resolved = sum(1 for s in staged if s["resolved"])
+    return {"ok": True, "source": body.source, "order_ref": body.order_ref, "cost_source": cost_source,
+            "staged": len(staged), "resolved": resolved, "unresolved": len(staged) - resolved, "errors": errors}
+
+
 def _selfcheck():
     # SKU from a Google-Form timestamp (both ISO and en-AU slash formats) → YYYYMMDDHHMMSS
     assert _sku_from_ts("2025-01-30T14:01:02.821Z") == "20250130140102"

app/main.py

@@ -110,6 +110,9 @@ _STARTUP_DDL = [
     "ALTER TABLE inventory ADD COLUMN IF NOT EXISTS cost_price numeric(10,2)",
     "ALTER TABLE inventory ADD COLUMN IF NOT EXISTS cost_source text",     # e.g. 'rarewaves #592619'
     "ALTER TABLE inventory ADD COLUMN IF NOT EXISTS condition_type text NOT NULL DEFAULT 'used'",  # 'new' | 'used'
+    # normalised-barcode lookup (DB barcodes stored inconsistently: '5 018775 901762' vs clean digits)
+    "CREATE INDEX IF NOT EXISTS disc_rel_id_barcode_norm ON disc_release_identifier "
+    "(regexp_replace(value,'[^0-9]','','g')) WHERE type='Barcode'",
 ]
 
 

site/admin.html

@@ -71,6 +71,7 @@
     <a data-v="wantlist">📝 Wantlist</a>
     <div class="sec">Catalog</div>
     <a data-v="intake">📥 Intake</a>
+    <a data-v="newstock">📦 New stock</a>
     <a data-v="heal">🩹 Heal</a>
     <a data-v="storemap">🗺️ Store map</a>
     <a data-v="crates">🗄️ Crates</a>
@@ -112,7 +113,7 @@ document.querySelectorAll('nav a[data-v]').forEach(a => a.onclick = () => go(a.d
 function go(v){
   if(ROLE!=='admin' && (v==='connections'||v==='staff'||v==='system')) v='dashboard';   // staff can't reach admin views
   document.querySelectorAll('nav a[data-v]').forEach(a => a.classList.toggle('on', a.dataset.v===v));
-  ({dashboard:vDash, inventory:vInv, orders:vOrders, customers:vCustomers, wantlist:vWantlist, intake:vIntake, heal:vHeal, storemap:vStoreMap, crates:vCrates, reports:vReports, connections:vConn, staff:vStaff, system:vSystem}[v])();
+  ({dashboard:vDash, inventory:vInv, orders:vOrders, customers:vCustomers, wantlist:vWantlist, intake:vIntake, newstock:vNewstock, heal:vHeal, storemap:vStoreMap, crates:vCrates, reports:vReports, connections:vConn, staff:vStaff, system:vSystem}[v])();
 }
 
 // ---------- Dashboard (fast tiles first, charts lazy) ----------
@@ -736,6 +737,70 @@ async function healRun(){
                        : '<div class="muted" style="margin-top:8px">All gaps closed. Re-scan to confirm.</div>');
 }
 
+// ---------- New stock (distro purchases → approve → publish) ----------
+async function vNewstock(){
+  $('#content').innerHTML = '<h2>New stock <span class="muted" style="font-size:13px">· approve distro purchases, then publish</span></h2>'
+    + '<div id="ns_bar" class="muted" style="margin:8px 0">loading…</div><div id="ns_body"></div>';
+  nsLoad();
+}
+async function nsLoad(){
+  const d = await api('/newstock');
+  if(!d.items.length){ $('#ns_bar').innerHTML='<div class="card" style="color:var(--ok)">✓ Nothing pending — scrape a distro order with the PRICEGOD extension and it lands here.</div>'; $('#ns_body').innerHTML=''; return; }
+  const resolved = d.items.filter(i=>i.release_id).length, priced = d.items.filter(i=>i.price!=null).length;
+  $('#ns_bar').innerHTML = `<b>${d.count}</b> copies pending · ${resolved} matched · ${priced} priced `
+    + `<button onclick="nsPublishAll()" style="margin-left:8px">🚀 Publish all matched &amp; priced</button>`;
+  const groups = {}; d.items.forEach(i=>{ (groups[i.cost_source]=groups[i.cost_source]||[]).push(i); });
+  $('#ns_body').innerHTML = Object.entries(groups).map(([src,items])=>
+    `<div class="card"><h3>${escapeH(src)} <span class="muted" style="font-weight:400">· ${items.length} copies</span></h3>
+     <div style="display:grid;grid-template-columns:repeat(auto-fill,minmax(310px,1fr));gap:10px">${items.map(nsCard).join('')}</div></div>`).join('');
+}
+function nsCard(it){
+  const cover = it.thumb ? `<img src="${escapeH(it.thumb)}" style="width:54px;height:54px;object-fit:cover;border-radius:5px;flex-shrink:0">`
+    : '<div style="width:54px;height:54px;background:var(--line);border-radius:5px;flex-shrink:0"></div>';
+  return `<div id="nsc_${it.sku}" style="display:flex;gap:9px;border:1px solid var(--line);border-radius:8px;padding:9px">${cover}
+    <div style="flex:1;min-width:0">
+      <div style="font-weight:600;font-size:13px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap">${escapeH(it.title||it.barcode||'?')}</div>
+      <div class="muted" style="font-size:11px">${escapeH(it.artist||'')}${it.year?' · '+it.year:''} · ${escapeH(it.barcode||'')}</div>
+      <div style="font-size:11px;margin:2px 0">${it.release_id?`<span style="color:var(--ok)">✓ rel ${it.release_id}</span>`:'<span style="color:var(--warn)">⚠ needs release</span>'} · cost ${money(it.cost_price)}</div>
+      <div style="display:flex;gap:5px;align-items:center;margin-top:4px">
+        <select class="ns_ct">${['new','used'].map(c=>`<option${c===it.condition_type?' selected':''}>${c}</option>`).join('')}</select>
+        <input class="ns_pr" type="number" step="0.01" placeholder="retail $" value="${it.price!=null?it.price:''}" style="width:78px">
+        <button onclick="nsPublish('${it.sku}')"${it.release_id?'':' disabled title="pick a release first"'}>✓ publish</button>
+        <button class="ghost" onclick="nsRepick('${it.sku}')">🔎</button>
+      </div>
+      <div class="ns_msg muted" style="font-size:11px;margin-top:3px"></div>
+    </div></div>`;
+}
+async function nsPublish(sku){
+  const c=$('#nsc_'+sku), price=parseFloat(c.querySelector('.ns_pr').value)||null, ct=c.querySelector('.ns_ct').value, msg=c.querySelector('.ns_msg');
+  if(price==null){ msg.innerHTML='<span style="color:var(--warn)">set a retail price first</span>'; return; }
+  msg.textContent='publishing…';
+  const r=await fetch('/admin/newstock/'+encodeURIComponent(sku),{method:'POST',headers:hdr(),body:JSON.stringify({price,condition_type:ct,publish:true})});
+  if(r.ok){ c.style.opacity=.4; msg.innerHTML='<span style="color:var(--ok)">✓ published & live</span>'; }
+  else msg.innerHTML='<span style="color:var(--warn)">failed</span>';
+}
+async function nsPublishAll(){
+  if(!confirm('Publish every matched copy that has a retail price?')) return;
+  const d=await fetch('/admin/newstock/publish-all',{method:'POST',headers:hdr()}).then(r=>r.json());
+  alert('Published '+d.published+' copies.'); nsLoad();
+}
+function nsRepick(sku){
+  const c=$('#nsc_'+sku);
+  c.innerHTML=`<div style="flex:1"><div style="display:flex;gap:5px"><input class="ns_q" placeholder="search artist / title" style="flex:1" onkeydown="if(event.key==='Enter')nsSearch('${sku}')"><button onclick="nsSearch('${sku}')">Search</button><button class="ghost" onclick="nsLoad()">✕</button></div><div class="ns_res" style="margin-top:6px"></div></div>`;
+  c.querySelector('.ns_q').focus();
+}
+async function nsSearch(sku){
+  const c=$('#nsc_'+sku), q=c.querySelector('.ns_q').value.trim(); if(!q) return;
+  const res=c.querySelector('.ns_res'); res.innerHTML='<span class="muted">searching…</span>';
+  const d=await api('/intake/search?q='+encodeURIComponent(q)); window['_nsres_'+sku]=d.items||[];
+  res.innerHTML=(d.items||[]).slice(0,8).map((r,idx)=>`<div onclick="nsSetRelease('${sku}',${idx})" style="cursor:pointer;padding:4px;border-bottom:1px solid var(--line);font-size:12px">${escapeH(r.artist||'')} – ${escapeH(r.title||'')} <span class="muted">${escapeH(r.year||'')} ${escapeH(r.country||'')}</span></div>`).join('')||'<span class="muted">no matches — try the barcode or fewer words</span>';
+}
+async function nsSetRelease(sku,idx){
+  const r=(window['_nsres_'+sku]||[])[idx]; if(!r) return;
+  await fetch('/admin/newstock/'+encodeURIComponent(sku),{method:'POST',headers:hdr(),body:JSON.stringify({release_id:r.release_id,title:(r.artist?r.artist+' – ':'')+(r.title||'')})});
+  nsLoad();
+}
+
 // ---------- Crates ----------
 async function vCrates(){
   const m = $('#content'); m.innerHTML = '<h2>Crates</h2><div class="muted">loading…</div>';