← back to Reid Witlin Onboarding
recover_tk10066.py
284 lines
#!/usr/bin/env python3
"""Execute Steve-approved TK-10066 recovery from an immutable manifest.
Dry-run is the default. Live mode archives exact duplicate DRAFTs, moves the
Pool-B side of each cross-identity collision to a fresh SKU, and repairs two
canonical links. Every operation is CAS-guarded, read back, and JSONL-ledgered.
"""
import argparse
import datetime as dt
import fcntl
import json
import os
import subprocess
import time
import create2
HERE = os.path.dirname(os.path.abspath(__file__))
LOCK = os.path.join(HERE, "verification", "tk10066-recovery.lock")
LEDGER = os.path.join(HERE, "verification", "tk10066-recovery-results.jsonl")
PRODUCT = """query($id:ID!){product(id:$id){id handle title status vendor variants(first:10){nodes{id sku price}}}}"""
SKU_LOOKUP = """query($query:String!){productVariants(first:20,query:$query){nodes{id sku product{id handle status}}}}"""
ARCHIVE = """mutation($input:ProductInput!){productUpdate(input:$input){product{id handle status}userErrors{field message}}}"""
RESKU = """mutation($pid:ID!,$variants:[ProductVariantsBulkInput!]!){productVariantsBulkUpdate(productId:$pid,variants:$variants){productVariants{id sku}userErrors{field message}}}"""
def now():
return dt.datetime.now(dt.timezone.utc).isoformat()
def q(value):
return "'" + str(value).replace("'", "''") + "'"
def gql_data(query, variables):
response = create2.gql(query, variables)
if response.get("errors"):
raise RuntimeError(response["errors"])
return response.get("data") or {}
def product(pid):
return gql_data(PRODUCT, {"id": pid}).get("product")
def sku_matches(sku):
connection = gql_data(SKU_LOOKUP, {"query": f"sku:{sku}"}).get("productVariants") or {}
return [node for node in connection.get("nodes") or [] if node.get("sku") == sku]
def psql(sql, tuples=False):
command = ["psql", "host=/tmp dbname=dw_unified", "-X", "-v", "ON_ERROR_STOP=1"]
if tuples:
command += ["-qAt", "-F", "\t"]
command += ["-c", sql]
result = subprocess.run(command, capture_output=True, text=True)
if result.returncode:
raise RuntimeError(result.stderr.strip() or result.stdout.strip())
return result.stdout.strip()
def record(kind, state, payload):
entry = {"ts": now(), "ticket": "TK-10066", "kind": kind, "state": state, **payload}
with open(LEDGER, "a") as fh:
fh.write(json.dumps(entry, sort_keys=True) + "\n")
fh.flush()
os.fsync(fh.fileno())
def archive_one(item, live):
archive_ids = item["archive_product_ids"]
if len(archive_ids) != 1:
raise RuntimeError(f"Expected one archive target for {item['sku']}")
pid = archive_ids[0]
expected = next(p for p in item["shopify_products"] if p["product_id"] == pid)
before = product(pid)
if not before or before["handle"] != expected["handle"] or before["vendor"] != "Architectural Fabrics":
raise RuntimeError(f"Archive identity drift for {pid}: {before}")
if before["status"] == "ARCHIVED":
record("archive", "already_applied", {"sku": item["sku"], "product_id": pid})
return
variants = before.get("variants", {}).get("nodes", [])
if before["status"] != "DRAFT" or item["sku"] not in {v.get("sku") for v in variants}:
raise RuntimeError(f"Archive CAS failed for {pid}: {before}")
if not live:
record("archive", "dry_run", {"sku": item["sku"], "product_id": pid, "before": before})
return
result = gql_data(ARCHIVE, {"input": {"id": pid, "status": "ARCHIVED"}})["productUpdate"]
if result.get("userErrors"):
raise RuntimeError(result["userErrors"])
after = product(pid)
if not after or after["status"] != "ARCHIVED":
raise RuntimeError(f"Archive read-back failed for {pid}: {after}")
record("archive", "applied", {"sku": item["sku"], "product_id": pid,
"before": before, "after": after})
def db_resku(item):
old, new = item["old_sku"], item["new_sku"]
mover, retain = item["move_identity"], item["retain_identity"]
pid, handle = mover["product_id"], item["move_product"]["handle"]
retain_pid = retain["product_id"]
retain_gid = retain_pid if retain_pid.startswith("gid://") else f"gid://shopify/Product/{retain_pid}"
retain_product = next(p for p in item["shopify_products"] if p["product_id"] == retain_gid)
retain_handle = retain_product["handle"]
sql = f"""
BEGIN;
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM rwltd_catalog WHERE id={int(mover['row_id'])}
AND mfr_sku={q(mover['mfr_sku'])} AND dw_sku IN ({q(old)},{q(new)})) THEN
RAISE EXCEPTION 'mover catalog CAS failed';
END IF;
IF NOT EXISTS (SELECT 1 FROM rwltd_catalog WHERE id={int(retain['row_id'])}
AND mfr_sku={q(retain['mfr_sku'])} AND dw_sku={q(old)}) THEN
RAISE EXCEPTION 'retain catalog CAS failed';
END IF;
END $$;
UPDATE rwltd_catalog SET dw_sku={q(new)}, shopify_product_id={q(pid)},
shopify_handle={q(handle)}, on_shopify=true, agent_updated_at=now()
WHERE id={int(mover['row_id'])} AND mfr_sku={q(mover['mfr_sku'])};
UPDATE rwltd_catalog SET shopify_product_id={q(retain_gid)},shopify_handle={q(retain_handle)},
on_shopify=true,agent_updated_at=now()
WHERE id={int(retain['row_id'])} AND mfr_sku={q(retain['mfr_sku'])} AND dw_sku={q(old)};
INSERT INTO dw_sku_registry
(dw_sku,vendor_prefix,vendor_name,mfr_sku,shopify_product_id,shopify_handle,status,updated_at)
VALUES ({q(old)},'DWKR','Architectural Fabrics',{q(retain['mfr_sku'])},
{q(retain_gid)},{q(retain_handle)},'draft',now())
ON CONFLICT (vendor_prefix,mfr_sku) DO UPDATE SET dw_sku=EXCLUDED.dw_sku,
shopify_product_id=EXCLUDED.shopify_product_id,shopify_handle=EXCLUDED.shopify_handle,
status='draft',updated_at=now();
INSERT INTO dw_sku_registry
(dw_sku,vendor_prefix,vendor_name,mfr_sku,shopify_product_id,shopify_handle,status,updated_at)
VALUES ({q(new)},'DWKR','Architectural Fabrics',{q(mover['mfr_sku'])},{q(pid)},{q(handle)},'draft',now())
ON CONFLICT (vendor_prefix,mfr_sku) DO UPDATE SET dw_sku=EXCLUDED.dw_sku,
shopify_product_id=EXCLUDED.shopify_product_id,shopify_handle=EXCLUDED.shopify_handle,
status='draft',updated_at=now();
COMMIT;
"""
psql(sql)
def verify_db_resku(item):
mover = item["move_identity"]
output = psql(
f"SELECT dw_sku,COALESCE(shopify_product_id,''),COALESCE(shopify_handle,'') "
f"FROM rwltd_catalog WHERE id={int(mover['row_id'])};", tuples=True)
if output != "\t".join([item["new_sku"], mover["product_id"], item["move_product"]["handle"]]):
raise RuntimeError(f"Catalog resku read-back failed: {output}")
registry = psql(
f"SELECT dw_sku,COALESCE(shopify_product_id,'') FROM dw_sku_registry "
f"WHERE vendor_prefix='DWKR' AND mfr_sku={q(mover['mfr_sku'])};", tuples=True)
if registry != "\t".join([item["new_sku"], mover["product_id"]]):
raise RuntimeError(f"Registry resku read-back failed: {registry}")
retain = item["retain_identity"]
retain_gid = retain["product_id"] if retain["product_id"].startswith("gid://") else \
f"gid://shopify/Product/{retain['product_id']}"
retain_product = next(p for p in item["shopify_products"] if p["product_id"] == retain_gid)
retain_check = psql(
f"SELECT dw_sku,COALESCE(shopify_product_id,''),COALESCE(shopify_handle,''),on_shopify::text "
f"FROM rwltd_catalog WHERE id={int(retain['row_id'])};", tuples=True)
expected_retain = "\t".join([item["old_sku"], retain_gid, retain_product["handle"], "true"])
if retain_check != expected_retain:
raise RuntimeError(f"Retain catalog read-back failed: {retain_check}")
def resku_one(item, live):
old_variant_sku, new_variant_sku = item["old_sku"] + "-Sample", item["new_sku"] + "-Sample"
expected = item["move_product"]
before = product(expected["product_id"])
if not before or before["handle"] != expected["handle"] or before["status"] != "DRAFT":
raise RuntimeError(f"Re-SKU identity/status drift: {before}")
variant = next((v for v in before["variants"]["nodes"] if v["id"] == expected["variant_id"]), None)
if not variant or variant["sku"] not in (old_variant_sku, new_variant_sku):
raise RuntimeError(f"Re-SKU variant CAS failed: {variant}")
conflicts = [v for v in sku_matches(new_variant_sku)
if v["product"]["id"] != expected["product_id"]]
if conflicts:
raise RuntimeError(f"Fresh SKU collision {new_variant_sku}: {conflicts}")
if not live:
record("resku", "dry_run", {"old_sku": item["old_sku"], "new_sku": item["new_sku"],
"product_id": expected["product_id"], "before": before})
return
changed_shopify = variant["sku"] == old_variant_sku
if changed_shopify:
result = gql_data(RESKU, {"pid": expected["product_id"], "variants": [{
"id": expected["variant_id"], "inventoryItem": {"sku": new_variant_sku}}]})["productVariantsBulkUpdate"]
if result.get("userErrors"):
raise RuntimeError(result["userErrors"])
after_shopify = product(expected["product_id"])
after_variant = next(v for v in after_shopify["variants"]["nodes"] if v["id"] == expected["variant_id"])
if after_variant["sku"] != new_variant_sku:
raise RuntimeError(f"Shopify re-SKU read-back failed: {after_variant}")
try:
db_resku(item)
verify_db_resku(item)
except Exception:
if changed_shopify:
rollback = gql_data(RESKU, {"pid": expected["product_id"], "variants": [{
"id": expected["variant_id"], "inventoryItem": {"sku": old_variant_sku}}]})["productVariantsBulkUpdate"]
record("resku", "db_failed_shopify_rollback", {"old_sku": item["old_sku"],
"new_sku": item["new_sku"], "product_id": expected["product_id"], "rollback": rollback})
raise
record("resku", "applied" if changed_shopify else "already_applied", {
"old_sku": item["old_sku"], "new_sku": item["new_sku"],
"product_id": expected["product_id"], "before": before, "after": after_shopify})
def link_one(item, live):
target, shopify = item["target"], item["shopify_product"]
if not target or not shopify or shopify["status"] != "DRAFT":
raise RuntimeError(f"Link identity/status invalid: {item}")
expected_variant = target["sku"] + "-Sample"
if expected_variant not in {v["sku"] for v in shopify["variants"]["nodes"]}:
raise RuntimeError(f"Link SKU mismatch for {item['handle']}")
if not live:
record("link", "dry_run", {"mfr_sku": target["mfr_sku"], "sku": target["sku"],
"product_id": shopify["id"]})
return
sql = f"""
BEGIN;
UPDATE rwltd_catalog SET shopify_product_id={q(shopify['id'])},shopify_handle={q(item['handle'])},
dw_sku={q(target['sku'])},on_shopify=true,agent_updated_at=now()
WHERE mfr_sku={q(target['mfr_sku'])} AND dw_sku={q(target['sku'])};
INSERT INTO dw_sku_registry
(dw_sku,vendor_prefix,vendor_name,mfr_sku,shopify_product_id,shopify_handle,status,updated_at)
VALUES ({q(target['sku'])},'DWKR','Architectural Fabrics',{q(target['mfr_sku'])},
{q(shopify['id'])},{q(item['handle'])},'draft',now())
ON CONFLICT (vendor_prefix,mfr_sku) DO UPDATE SET
shopify_product_id=EXCLUDED.shopify_product_id,shopify_handle=EXCLUDED.shopify_handle,
status='draft',updated_at=now();
COMMIT;
"""
psql(sql)
check = psql(f"SELECT COALESCE(shopify_product_id,''),COALESCE(shopify_handle,'') "
f"FROM rwltd_catalog WHERE mfr_sku={q(target['mfr_sku'])} AND dw_sku={q(target['sku'])};",
tuples=True)
if check != "\t".join([shopify["id"], item["handle"]]):
raise RuntimeError(f"Link read-back failed: {check}")
record("link", "applied", {"mfr_sku": target["mfr_sku"], "sku": target["sku"],
"product_id": shopify["id"]})
def main():
parser = argparse.ArgumentParser()
parser.add_argument("manifest")
parser.add_argument("--apply", action="store_true")
parser.add_argument("--limit-archive", type=int, default=0)
parser.add_argument("--limit-resku", type=int, default=0)
parser.add_argument("--limit-links", type=int, default=0)
args = parser.parse_args()
manifest_path = os.path.realpath(args.manifest)
if "/verification/recovery-manifests/" not in manifest_path:
raise SystemExit("Refusing mutable/non-verification manifest")
manifest = json.load(open(manifest_path))
if manifest.get("schema") != "tk-10066-recovery-manifest/v1" or not all(manifest["assertions"].values()):
raise SystemExit("Manifest schema/assertions invalid")
os.makedirs(os.path.dirname(LOCK), exist_ok=True)
with open(LOCK, "w") as lock:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
raise SystemExit("Another TK-10066 recovery writer holds the lock")
archive = manifest["same_identity"][:args.limit_archive or None]
resku = manifest["cross_identity"][:args.limit_resku or None]
links = manifest["handle_repairs"][:args.limit_links or None]
record("run", "start", {"mode": "apply" if args.apply else "dry_run",
"manifest": manifest_path, "counts": [len(archive), len(resku), len(links)]})
for item in archive:
archive_one(item, args.apply); time.sleep(0.25 if args.apply else 0)
for item in resku:
resku_one(item, args.apply); time.sleep(0.25 if args.apply else 0)
for item in links:
link_one(item, args.apply)
record("run", "complete", {"mode": "apply" if args.apply else "dry_run",
"counts": [len(archive), len(resku), len(links)]})
print(json.dumps({"ok": True, "mode": "apply" if args.apply else "dry_run",
"archive": len(archive), "resku": len(resku), "links": len(links)}))
if __name__ == "__main__":
main()