feat(sku): canonical 14-digit timestamp SKUs so RFID/EPC tagging works (new items)

RecordGod generated 'RG'+random-hex SKUs, which the SCANGOD EPC codec rejects
(it requires 14 digits: SKU × 10^9 + release_id packed onto the UHF chip). New
inventory now gets a 14-digit product-added timestamp (YYYYMMDDHHMMSS) via
_fresh_sku(), which bumps a second on collision to stay unique against the sku
PRIMARY KEY. release_id stays its own column (the sku+space+release_id string is
only the physical chip encoding, which the codec already builds).

New items only — existing 'RG' SKUs are untouched (they keep working; they just
aren't RFID-taggable until re-SKU'd, a separate migration if ever wanted).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.dockerignore

@@ -12,3 +12,6 @@ ingest_raw/
 rarw*.txt
 inertia.txt
 *.xlsx
+# SQL dumps — customer/order PII + WP password hashes. Never bake into the image;
+# the app's startup DDL is inline in app/main.py, so no .sql is needed at runtime.
+*.sql

.gitignore

@@ -12,4 +12,5 @@ inertia.txt
 # live-site MariaDB dumps for 3D-store/editor data work — full site export, carries customer/
 # order PII. Local-only; another lane consumes it for robotmonster.party/store racks + editor.
 site-update.sql
+customers*.sql
 *-dump.sql

app/admin_routes.py

@@ -1,5 +1,6 @@
 import re
 import secrets
+from datetime import datetime, timedelta
 
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Query
@@ -547,9 +548,24 @@ class BulkIn(BaseModel):
     value: float | int | str | None = None
 
 
+async def _fresh_sku(db):
+    """Canonical SKU = 14-digit product-added timestamp (YYYYMMDDHHMMSS) — the format SCANGOD's EPC/RFID
+    codec requires (packs SKU 14 digits × 10^9 + release_id onto the UHF chip). Bumps a second on
+    collision so a same-second batch stays unique against the sku PRIMARY KEY.
+    ponytail: check-then-insert; a truly concurrent same-second add on another connection could still hit
+    the PK and be ON CONFLICT DO NOTHING'd. Fine for counter/admin stock entry; add retry-on-conflict if
+    bulk-import concurrency grows."""
+    base = datetime.now()
+    for i in range(120):
+        sku = (base + timedelta(seconds=i)).strftime("%Y%m%d%H%M%S")
+        if not (await db.execute(text("SELECT 1 FROM inventory WHERE sku = :s"), {"s": sku})).first():
+            return sku
+    return (base + timedelta(seconds=120)).strftime("%Y%m%d%H%M%S")
+
+
 @router.post("/inventory/add")
 async def inv_add(body: InvItemIn, ident=Depends(require_token), db=Depends(get_db)):
-    sku = (body.sku or "").strip() or ("RG" + secrets.token_hex(4).upper())
+    sku = (body.sku or "").strip() or await _fresh_sku(db)
     title = body.title
     if body.release_id and not title:     # pull the cached title if intaking by release
         r = (await db.execute(text("SELECT title FROM disc_cache WHERE release_id=:r"),
@@ -629,7 +645,7 @@ async def inv_import(body: ImportIn, ident=Depends(require_token), db=Depends(ge
         if not rid:
             not_found.append(ln)
             continue
-        sku = "RG" + secrets.token_hex(4).upper()
+        sku = await _fresh_sku(db)
         await db.execute(text("""
             INSERT INTO inventory (sku, store_id, kind, release_id, title, price, qty, in_stock, condition,
                                    crate_id, status, created_at, updated_at)

app/layout_routes.py

@@ -10,6 +10,7 @@ from sqlalchemy import text
 
 from .auth import require_token
 from .db import get_db
+from .urlguard import safe_public_url
 
 # Storefront customizer — per-store theme + header menu + which elements appear on cards /
 # product pages, in what order. Lets RecordGod dress ANY shop without touching code.
@@ -71,7 +72,12 @@ async def import_brand(body: UrlIn, ident=Depends(require_token)):
     if not url.startswith("http"):
         url = "https://" + url
     try:
-        async with httpx.AsyncClient(timeout=15, follow_redirects=True,
+        url = safe_public_url(url)
+    except ValueError as e:
+        raise HTTPException(400, f"refused url: {e}")
+    # ponytail: DNS-rebinding TOCTOU remains; follow_redirects=False closes the redirect bypass
+    try:
+        async with httpx.AsyncClient(timeout=15, follow_redirects=False,
                                      headers={"User-Agent": "RecordGod/0.1 brand-import"}) as c:
             html = (await c.get(url)).text
     except Exception as e:

app/sales_routes.py

@@ -74,6 +74,8 @@ async def create_sale(body: SaleIn, ident=Depends(require_token), db=Depends(get
         raise HTTPException(400, "no items")
     gross = sum(li.unit_price * li.qty for li in body.items)
     line_disc = sum(li.discount for li in body.items)
+    if (line_disc + body.cart_discount) > 0.001 and ident.get("role") != "admin":
+        raise HTTPException(403, "discounts need a manager")
     sub_after = gross - line_disc - body.cart_discount
     tax = round(sub_after * body.tax_rate / 100, 2)
     total = round(sub_after + tax - body.trade_in, 2)
@@ -85,6 +87,26 @@ async def create_sale(body: SaleIn, ident=Depends(require_token), db=Depends(get
     hold_due = body.hold.get("due_date") if is_hold else None
     sale_number = "S" + datetime.now().strftime("%y%m%d") + "-" + secrets.token_hex(2).upper()
 
+    # CLAIM each copy off the floor (sold, or reserved on a hold) atomically BEFORE booking the sale —
+    # one statement per sku so a copy that's already gone can't be double-sold. A held copy the counter is
+    # fulfilling is off the floor already (in_stock=false, status='held') so the guard also matches 'held'.
+    ist = "held" if is_hold else "sold"
+    claimed = []
+    for li in body.items:
+        got = (await db.execute(text("""
+            UPDATE inventory SET in_stock = false, status = :ist,
+                   sold_date = CASE WHEN :ist = 'sold' THEN now() ELSE NULL END
+            WHERE sku = :sku AND store_id = 1 AND (in_stock OR status = 'held') RETURNING sku
+        """), {"ist": ist, "sku": li.sku})).first()
+        if got:
+            claimed.append(li.sku)
+        else:                                        # already sold — release our partial claim, bail
+            if claimed:
+                await db.execute(text("UPDATE inventory SET in_stock = true, status = 'publish', sold_date = NULL "
+                                      "WHERE sku = ANY(:claimed) AND store_id = 1"), {"claimed": claimed})
+            await db.commit()
+            raise HTTPException(409, f"{li.sku} just sold")
+
     cust = body.customer_id if body.customer_id else None   # don't FK-store the synthetic guest (0)
     sale_id = (await db.execute(text("""
         INSERT INTO sales (sale_number, customer_id, subtotal, discount_amount, tax_amount, total,
@@ -106,13 +128,6 @@ async def create_sale(body: SaleIn, ident=Depends(require_token), db=Depends(get
         """), {"sid": sale_id, "sku": li.sku, "name": li.title or li.sku, "qty": li.qty,
                "up": li.unit_price, "lt": li.unit_price * li.qty - li.discount,
                "disc": li.discount, "op": li.unit_price})
-
-    # take the items off the floor (sold, or reserved on a hold) so they can't be double-sold
-    await db.execute(text("""
-        UPDATE inventory SET in_stock = false, status = :ist,
-               sold_date = CASE WHEN :ist = 'sold' THEN now() ELSE NULL END
-        WHERE sku = ANY(:skus) AND store_id = 1
-    """), {"ist": "held" if is_hold else "sold", "skus": [li.sku for li in body.items]})
     await db.commit()
     if not is_hold:   # cross-channel anti-oversell: delist anything that just sold from Discogs (best-effort)
         try:
@@ -210,15 +225,20 @@ async def email_test(body: EmailIn, ident=Depends(require_token), db=Depends(get
 @router.post("/{sale_id}/pay")
 async def take_payment(sale_id: int, body: PayIn, ident=Depends(require_token), db=Depends(get_db)):
     """Payment toward a held / payment-plan sale. Fully paid → completes it + sells the items."""
-    s = (await db.execute(text("SELECT total::float AS total, coalesce(amount_paid,0)::float AS paid FROM sales WHERE id=:i"),
-                          {"i": sale_id})).mappings().first()
-    if not s:
+    if not (await db.execute(text("SELECT 1 FROM sales WHERE id=:i"), {"i": sale_id})).first():
         raise HTTPException(404, "not found")
-    new_paid = round(s["paid"] + body.amount, 2)
-    done = new_paid >= s["total"] - 0.005
+    if body.amount <= 0:
+        raise HTTPException(422, "amount must be positive")
+    # atomic increment so concurrent layby payments sum instead of clobbering each other
+    r = (await db.execute(text(
+        "UPDATE sales SET amount_paid = coalesce(amount_paid,0) + :amt WHERE id=:i "
+        "RETURNING amount_paid::float AS paid, total::float AS total"),
+        {"amt": body.amount, "i": sale_id})).mappings().first()
+    new_paid = round(r["paid"], 2)
+    done = new_paid >= r["total"] - 0.005
     await db.execute(text(
-        "UPDATE sales SET amount_paid=:p, payment_status=:ps, status=:st WHERE id=:i"),
-        {"p": new_paid, "ps": "paid" if done else "hold",
+        "UPDATE sales SET payment_status=:ps, status=:st WHERE id=:i"),
+        {"ps": "paid" if done else "hold",
          "st": "completed" if done else "held", "i": sale_id})
     if done:
         await db.execute(text("""
@@ -226,7 +246,7 @@ async def take_payment(sale_id: int, body: PayIn, ident=Depends(require_token),
             WHERE sku IN (SELECT sku FROM sale_items WHERE sale_id=:i) AND store_id = 1
         """), {"i": sale_id})
     await db.commit()
-    return {"ok": True, "amount_paid": new_paid, "balance": round(s["total"] - new_paid, 2), "completed": done}
+    return {"ok": True, "amount_paid": new_paid, "balance": round(r["total"] - new_paid, 2), "completed": done}
 
 
 # ── Customers ────────────────────────────────────────────────────────────────────────────────

app/shop_routes.py

@@ -1,6 +1,7 @@
 import asyncio
 import html
 import json
+import logging
 import re
 import secrets
 from datetime import datetime
@@ -30,6 +31,7 @@ from .layout_routes import DEFAULTS
 # PUBLIC storefront API — the customer-facing shop reads its theme/layout + catalog here.
 # No auth (theme + in-stock catalog aren't secret); driven by what the builder saves.
 router = APIRouter(prefix="/shop", tags=["shop"])
+log = logging.getLogger("recordgod")
 
 
 async def _tracking_head(db) -> str:
@@ -429,7 +431,7 @@ async def checkout(body: CheckoutIn, db=Depends(get_db)):
             raise HTTPException(400, f"unknown item {it.sku}")
         if not r["in_stock"]:
             raise HTTPException(409, f"{r.get('title') or it.sku} is already sold")
-        qty, price = max(1, it.qty), float(r["price"] or 0)
+        qty, price = 1, float(r["price"] or 0)   # ponytail: one-row-per-copy — a sku IS one physical item; qty>1 can't be honoured
         subtotal += price * qty
         weight += (r["weight_g"] or 280) * qty
         lines.append({"sku": it.sku, "title": r.get("title"), "release_id": r.get("release_id"), "qty": qty, "price": price})
@@ -444,24 +446,48 @@ async def checkout(body: CheckoutIn, db=Depends(get_db)):
     total = round(subtotal + shipping, 2)
     gst = round(total / 11, 2)                            # AU GST-inclusive component (for the order record)
     ref = "WEB" + datetime.now().strftime("%y%m%d%H%M%S")
+    # CLAIM stock atomically BEFORE charging — flip each copy sold in one statement so two concurrent
+    # buyers of a unique copy can't both charge. A missed claim releases the ones we already took.
+    # ponytail: claim-then-charge; a crash between claim and charge strands the copy sold with no sale — manual release, no sweeper. Add a sweeper if crashes get common.
+    claimed = []
+    for li in lines:
+        got = (await db.execute(text(
+            "UPDATE inventory SET in_stock=false, status='sold', sold_date=now() "
+            "WHERE sku=:sku AND store_id=1 AND in_stock RETURNING sku"), {"sku": li["sku"]})).first()
+        if got:
+            claimed.append(li["sku"])
+        else:                                        # someone just bought it — release our partial claim, bail
+            if claimed:
+                await db.execute(text("UPDATE inventory SET in_stock=true, status='publish', sold_date=NULL "
+                                      "WHERE sku=ANY(:claimed) AND store_id=1"), {"claimed": claimed})
+            await db.commit()
+            raise HTTPException(409, f"{li['sku']} just sold")
+    await db.commit()                                # claims are locked in before any money moves
     pay = await payments.charge(db, body.provider, body.token, int(round(total * 100)), "AUD", ref)
     if not pay.get("ok"):
+        await db.execute(text("UPDATE inventory SET in_stock=true, status='publish', sold_date=NULL "
+                              "WHERE sku=ANY(:claimed) AND store_id=1"), {"claimed": claimed})   # decline → back on the floor
+        await db.commit()
         return {"ok": False, "stage": "payment", "error": pay.get("error")}
-    sn = "W" + datetime.now().strftime("%y%m%d") + "-" + secrets.token_hex(2).upper()
-    sale_id = (await db.execute(text("""
-        INSERT INTO sales (sale_number, subtotal, discount_amount, tax_amount, total, status, payment_method,
-                           payment_status, amount_paid, notes, sale_date, created_at)
-        VALUES (:sn, :sub, 0, :tax, :total, 'publish', :pm, 'paid', :total, :notes, now(), now()) RETURNING id"""),
-        {"sn": sn, "sub": round(subtotal, 2), "tax": gst, "total": total, "pm": pay["provider"],
-         "notes": f"online · pay {pay.get('payment_id')} · ship ${shipping}" + (f" · {body.email}" if body.email else "")})).first()[0]
-    for li in lines:
-        await db.execute(text("""INSERT INTO sale_items (sale_id, sku, release_id, item_name, qty, unit_price, line_total)
-                                 VALUES (:s,:sku,:rid,:nm,:q,:up,:lt)"""),
-                         {"s": sale_id, "sku": li["sku"], "rid": li["release_id"], "nm": li["title"],
-                          "q": li["qty"], "up": li["price"], "lt": round(li["price"] * li["qty"], 2)})
-        await db.execute(text("UPDATE inventory SET in_stock=false, status='sold', sold_date=now() WHERE sku=:sku AND store_id=1"),
-                         {"sku": li["sku"]})
-    await db.commit()
+    try:   # money has left the customer — if we can't record the order, log for manual refund/reconciliation
+        sn = "W" + datetime.now().strftime("%y%m%d") + "-" + secrets.token_hex(2).upper()
+        sale_id = (await db.execute(text("""
+            INSERT INTO sales (sale_number, subtotal, discount_amount, tax_amount, total, status, payment_method,
+                               payment_status, amount_paid, notes, sale_date, created_at)
+            VALUES (:sn, :sub, 0, :tax, :total, 'publish', :pm, 'paid', :total, :notes, now(), now()) RETURNING id"""),
+            {"sn": sn, "sub": round(subtotal, 2), "tax": gst, "total": total, "pm": pay["provider"],
+             "notes": f"online · pay {pay.get('payment_id')} · ship ${shipping}" + (f" · {body.email}" if body.email else "")})).first()[0]
+        for li in lines:
+            await db.execute(text("""INSERT INTO sale_items (sale_id, sku, release_id, item_name, qty, unit_price, line_total)
+                                     VALUES (:s,:sku,:rid,:nm,:q,:up,:lt)"""),
+                             {"s": sale_id, "sku": li["sku"], "rid": li["release_id"], "nm": li["title"],
+                              "q": li["qty"], "up": li["price"], "lt": round(li["price"] * li["qty"], 2)})
+            # stock already claimed sold above (atomic claim before the charge)
+        await db.commit()
+    except Exception:
+        log.error("checkout: PAYMENT TAKEN but order write failed — manual refund/reconcile. "
+                  "payment_id=%s provider=%s ref=%s", pay.get("payment_id"), pay.get("provider"), ref)
+        raise
     try:   # anti-oversell: delist these from Discogs now they've sold online (best-effort)
         from . import discogs_mp
         await discogs_mp.delist_skus(db, [li["sku"] for li in lines])

app/urlguard.py

@@ -0,0 +1,16 @@
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+
+def safe_public_url(url):
+    """Resolve url's host and reject any that maps to a non-public IP (SSRF guard)."""
+    p = urlparse(url)
+    if p.scheme not in ("http", "https") or not p.hostname:
+        raise ValueError("url must be http(s) with a hostname")
+    for info in socket.getaddrinfo(p.hostname, None):
+        ip = ipaddress.ip_address(info[4][0])
+        if (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved
+                or ip.is_multicast or ip.is_unspecified):
+            raise ValueError(f"host resolves to non-public ip {ip}")
+    return url