"""Synthetic JSONL/SQLite laboratory. No HTTP call, Shopify shop or ERP write.
Run with Python 3: python3 import_jsonl_shopify.py
The operation metadata and file are fixtures, not Shopify responses.
"""
import copy
import hashlib
import json
import random
import sqlite3


def gid(kind, number):
    return f"gid://shopify/{kind}/{number}"


ROWS = [
    {"id": gid("Product", 100), "__typename": "Product", "title": "Chemise"},
    {"id": gid("ProductVariant", 101), "__typename": "ProductVariant", "sku": "C-S", "__parentId": gid("Product", 100)},
    {"id": gid("ProductVariant", 102), "__typename": "ProductVariant", "sku": "C-M", "__parentId": gid("Product", 100)},
    {"id": gid("Product", 200), "__typename": "Product", "title": "Gants"},
    {"id": gid("ProductVariant", 201), "__typename": "ProductVariant", "sku": "G-S", "__parentId": gid("Product", 200)},
    {"id": gid("Product", 300), "__typename": "Product", "title": "Bottes"},
    {"id": gid("ProductVariant", 301), "__typename": "ProductVariant", "sku": "B-40", "__parentId": gid("Product", 300)},
    {"id": gid("ProductVariant", 302), "__typename": "ProductVariant", "sku": "B-41", "__parentId": gid("Product", 300)},
    {"id": gid("ProductVariant", 303), "__typename": "ProductVariant", "sku": "B-42", "__parentId": gid("Product", 300)},
]
EXPECTED = {gid("ProductVariant", v): gid("Product", p) for v, p in [(101, 100), (102, 100), (201, 200), (301, 300), (302, 300), (303, 300)]}
REQUEST = {"shop": "boutique-a.example", "operation": gid("BulkOperation", 7), "queryHash": "query-products-variants-v1"}
META = {**REQUEST, "type": "QUERY", "status": "COMPLETED", "errorCode": None, "objectCount": "9", "rootObjectCount": "3", "url": "https://download.example.invalid/result.jsonl", "partialDataUrl": None}


def encode(rows):
    return ("\n".join(json.dumps(r, ensure_ascii=False) for r in rows) + "\n").encode("utf-8")


def no_duplicate_keys(pairs):
    obj = {}
    for key, value in pairs:
        if key in obj:
            raise ValueError("duplicate JSON key")
        obj[key] = value
    return obj


def reject_constant(value):
    raise ValueError("non-JSON constant")


def database():
    db = sqlite3.connect(":memory:")
    db.executescript("""
      CREATE TABLE active(shop TEXT, id TEXT, kind TEXT, parent TEXT, payload TEXT, PRIMARY KEY(shop,id));
      CREATE TABLE stage(id TEXT PRIMARY KEY, kind TEXT, parent TEXT, payload TEXT);
      CREATE TABLE receipts(shop TEXT, operation TEXT, query_hash TEXT, sha TEXT, PRIMARY KEY(shop,operation));
      INSERT INTO active VALUES('boutique-a.example','old-A','Product',NULL,'{}');
      INSERT INTO active VALUES('boutique-b.example','old-B','Product',NULL,'{}');
    """)
    return db


def state(db):
    return db.execute("SELECT * FROM active ORDER BY shop,id").fetchall()


def promote(db, request, meta, payload):
    # Local association checks, not fields asserted to be present in Shopify's response.
    if any(meta.get(k) != request[k] for k in ("shop", "operation", "queryHash")):
        raise ValueError("wrong export context")
    if meta["type"] != "QUERY" or meta["status"] != "COMPLETED" or meta["errorCode"] is not None:
        raise ValueError("export not complete")
    if meta.get("partialDataUrl"):
        raise ValueError("partial result not a replacement")
    if meta["objectCount"] == "0" and meta["rootObjectCount"] == "0" and meta["url"] is None:
        return "EMPTY_REVIEW"  # Valid empty-export shape; this pilot does not erase an existing catalog.
    if not meta["url"]:
        raise ValueError("missing result URL")
    digest = hashlib.sha256(payload).hexdigest()  # Local artifact fingerprint, not a Shopify checksum.
    previous = db.execute("SELECT query_hash,sha FROM receipts WHERE shop=? AND operation=?", (request["shop"],request["operation"])).fetchone()
    if previous:
        if previous != (request["queryHash"],digest):
            raise ValueError("same operation, changed artifact")
        return "ALREADY_PROMOTED"
    text = payload.decode("utf-8")
    if text.startswith("\ufeff"):
        raise ValueError("BOM forbidden")
    with db:
        db.execute("DELETE FROM stage")
        for number, line in enumerate(text.splitlines(), 1):
            obj = json.loads(line, object_pairs_hook=no_duplicate_keys, parse_constant=reject_constant)
            if not isinstance(obj, dict):
                raise ValueError(f"line {number}: object expected for this query")
            kind, identity = obj.get("__typename"), obj.get("id")
            if kind not in ("Product", "ProductVariant") or not isinstance(identity, str) or not identity.startswith(f"gid://shopify/{kind}/"):
                raise ValueError("unexpected type or identity")
            parent = obj.get("__parentId")
            if kind == "Product" and parent is not None:
                raise ValueError("unexpected parent on root")
            if kind == "ProductVariant" and (not isinstance(parent,str) or not parent.startswith("gid://shopify/Product/")):
                raise ValueError("missing product identity")
            db.execute("INSERT INTO stage VALUES(?,?,?,?)", (identity, kind, parent, json.dumps(obj,sort_keys=True)))
        total = db.execute("SELECT count(*) FROM stage").fetchone()[0]
        roots = db.execute("SELECT count(*) FROM stage WHERE kind='Product'").fetchone()[0]
        if total != int(meta["objectCount"]) or roots != int(meta["rootObjectCount"]):
            raise ValueError("counts disagree for this two-level query")
        orphan = db.execute("SELECT count(*) FROM stage v LEFT JOIN stage p ON v.parent=p.id AND p.kind='Product' WHERE v.kind='ProductVariant' AND p.id IS NULL").fetchone()[0]
        if orphan:
            raise ValueError("orphan variants")
        db.execute("DELETE FROM active WHERE shop=?", (request["shop"],))
        db.execute("INSERT INTO active SELECT ?,id,kind,parent,payload FROM stage", (request["shop"],))
        db.execute("INSERT INTO receipts VALUES(?,?,?,?)", (request["shop"], request["operation"], request["queryHash"],digest))
    return "PROMOTED"


def main():
    checks = 0
    def check(actual, expected):
        nonlocal checks
        assert actual == expected, (actual, expected)
        checks += 1

    orders=[ROWS,list(reversed(ROWS)),ROWS[3:]+ROWS[:3]]
    shuffled=copy.deepcopy(ROWS);random.Random(7).shuffle(shuffled);orders.append(shuffled)
    for rows in orders:
        db=database();payload=encode(rows)
        check(promote(db,REQUEST,META,payload),"PROMOTED")
        check(dict(db.execute("SELECT id,parent FROM active WHERE shop=? AND kind='ProductVariant'",(REQUEST['shop'],))),EXPECTED)
        check(db.execute("SELECT count(*) FROM active WHERE shop=? AND kind='Product'",(REQUEST['shop'],)).fetchone()[0],3)
        check(db.execute("SELECT id FROM active WHERE shop='boutique-b.example'").fetchone()[0],"old-B")
        before=state(db);check(promote(db,REQUEST,META,payload),"ALREADY_PROMOTED");check(state(db),before)
        db.close()
    failures=[]
    def refused(name, meta=META, payload=None):
        db=database();before=state(db)
        try: promote(db,REQUEST,meta,encode(ROWS) if payload is None else payload)
        except (ValueError, sqlite3.IntegrityError): pass
        else: raise AssertionError(name+" was accepted")
        check(state(db),before);check(db.execute('SELECT count(*) FROM receipts').fetchone()[0],0)
        failures.append(name);db.close()
    for key,value in [('status','FAILED'),('status','RUNNING'),('type','MUTATION'),('shop','boutique-b.example'),('operation',gid('BulkOperation',8)),('queryHash','other-query'),('objectCount','8'),('rootObjectCount','9'),('partialDataUrl','partial-fixture'),('url',None)]:
        refused(key+'='+str(value),{**META,key:value})
    refused('truncated JSON',payload=encode(ROWS)[:-7])
    refused('complete lines missing',payload=encode(ROWS[:-1]))
    refused('duplicate ID',payload=encode(ROWS+[ROWS[1]]))
    orphan=copy.deepcopy(ROWS);orphan[1]['__parentId']=gid('Product',999);refused('orphan',payload=encode(orphan))
    missing=copy.deepcopy(ROWS);missing[1].pop('__parentId');refused('missing parent',payload=encode(missing))
    refused('blank line',payload=encode(ROWS)+b'\n')
    refused('null row',payload=b'null\n'+encode(ROWS))
    refused('BOM',payload=b'\xef\xbb\xbf'+encode(ROWS))
    refused('duplicate JSON key',payload=b'{"id":"a","id":"b"}\n')
    db=database();before=state(db);check(promote(db,REQUEST,{**META,'objectCount':'0','rootObjectCount':'0','url':None},b''),'EMPTY_REVIEW');check(state(db),before)
    # Negative witness: attaching every variant to the latest preceding root.
    current=None;bad={}
    for row in orders[3]:
        if row['__typename']=='Product': current=row['id']
        else: bad[row['id']]=current
    check(bad==EXPECTED,False)
    db=database();promote(db,REQUEST,META,encode(ROWS));before=state(db)
    try: promote(db,REQUEST,META,encode(list(reversed(ROWS))))
    except ValueError: check(state(db),before)
    else: raise AssertionError('changed artifact accepted')
    check(40*3,120)
    print(json.dumps({'checks':checks,'permutations':4,'rejectedCases':failures,'roots':3,'variants':6,'objects':9,'expectedParents':EXPECTED,'badLastParent':bad,'emptyExport':'review without replacement','networkCalls':0},ensure_ascii=False,indent=2))

if __name__ == '__main__':
    main()