Update app/db.py

app/db.py

@@ -100,27 +100,54 @@ def insert_records(conn, table_name: str, records: List[Dict[str, Any]]):
 # ------------------------------------------------------------
 def bootstrap(org_id: str, payload: Dict[str, Any]):
     """
-    Accepts:
-      { orgId: "real-org", rows: [ {...}, {...} ] }
-    or any legacy shape.
+    Unified ingestion entrypoint. Accepts:
+      - {"orgId": "...", "rows": [...]}
+      - {"orgId": "...", "data": [...]}
+      - {"orgId": "...", "tables": {...}}
+      - legacy shapes (list of rows)
+    Stores the full payload into main.raw_rows (for audit), then inserts rows into per-table tables.
     """
     conn = get_conn(org_id)
     conn.execute("CREATE SCHEMA IF NOT EXISTS main")
     ensure_raw_table(conn)
 
-    # ---  adapt n8n shape  ---
-    if "rows" in payload and "orgId" in payload:
-        rows = payload["rows"]          # your exact field
-        org_id = payload["orgId"]       # override param
-    else:
-        rows = payload                  # fallback
-
-    # lineage
-    conn.execute("INSERT INTO main.raw_rows (row_data) VALUES (?)", (json.dumps(payload),))
-
-    # ingest
-    if isinstance(rows, list) and rows:
+    # Normalize shapes
+    rows = None
+    if isinstance(payload, dict) and "orgId" in payload:
+        # typical n8n shape
+        if "rows" in payload and isinstance(payload["rows"], list):
+            rows = payload["rows"]
+        elif "data" in payload and isinstance(payload["data"], list):
+            rows = payload["data"]
+        elif "tables" in payload and isinstance(payload["tables"], dict):
+            # multi-table: write each table separately below
+            # but still store entire payload for audit
+            pass
+    elif isinstance(payload, list):
+        rows = payload
+
+    # lineage / audit: only insert a JSON string (non-empty)
+    try:
+        raw_json = json.dumps(payload)
+        if raw_json and raw_json != "null" and raw_json != "[]":
+            conn.execute("INSERT INTO main.raw_rows (row_data) VALUES (?)", (raw_json,))
+    except Exception as e:
+        print(f"[db] ⚠️ failed to store raw payload for auditing: {e}")
+
+    # ingest single-list rows (common)
+    if rows:
         ensure_table(conn, "sales", rows[0])
         insert_records(conn, "sales", rows)
+    else:
+        # multi-table handler
+        if isinstance(payload, dict) and "tables" in payload:
+            for tname, recs in payload["tables"].items():
+                if not isinstance(recs, list) or not recs:
+                    continue
+                # sanitize table name (basic)
+                safe_table = "".join(c for c in tname.lower() if c.isalnum() or c == "_")
+                ensure_table(conn, safe_table, recs[0])
+                insert_records(conn, safe_table, recs)
 
     conn.close()
+