Count I-confluence in your own schema

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Count I-confluence in your own schema.

Lists every constraint of a schema, marks each as invariant-confluent (safe
to check on each replica alone) or not, names what would enforce the others,
and for each bound computes the escrow rate. The schema below is a small shop;
replace SCHEMA with your own constraints.
"""
import argparse, json

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--replicas", type=int, default=3, help="Replicas that accept writes (for the escrow split).")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

# (table, constraint, kind, detail). kinds: not_null, check_row, unique_generated, unique_chosen,
# fk_insert_only, fk_with_deletes, lower_bound (with decrements), upper_bound (with increments)
SCHEMA = [
    ("customers", "id PRIMARY KEY (uuid)", "unique_generated", {}),
    ("customers", "email UNIQUE", "unique_chosen", {}),
    ("customers", "email NOT NULL", "not_null", {}),
    ("customers", "country CHECK (length = 2)", "check_row", {}),
    ("products", "id PRIMARY KEY (uuid)", "unique_generated", {}),
    ("products", "sku UNIQUE", "unique_chosen", {}),
    ("products", "price CHECK (price > 0)", "check_row", {}),
    ("products", "name NOT NULL", "not_null", {}),
    ("stock", "on_hand >= 0", "lower_bound", {"slack": 900, "use_per_s": 2.5}),
    ("orders", "id PRIMARY KEY (uuid)", "unique_generated", {}),
    ("orders", "customer_id REFERENCES customers (customers are never deleted)", "fk_insert_only", {}),
    ("orders", "status CHECK (in placed, paid, shipped)", "check_row", {}),
    ("orders", "total CHECK (total >= 0)", "check_row", {}),
    ("order_lines", "order_id REFERENCES orders ON DELETE CASCADE", "fk_with_deletes", {}),
    ("order_lines", "product_id REFERENCES products (products are never deleted)", "fk_insert_only", {}),
    ("order_lines", "qty CHECK (qty > 0)", "check_row", {}),
    ("wallets", "balance >= 0", "lower_bound", {"slack": 50_000, "use_per_s": 12}),
    ("coupons", "redemptions <= max_uses", "upper_bound", {"slack": 500, "use_per_s": 0.4}),
    ("coupons", "code UNIQUE", "unique_chosen", {}),
    ("sessions", "token NOT NULL", "not_null", {}),
    ("sessions", "user_id REFERENCES customers", "fk_insert_only", {}),
    ("reviews", "rating CHECK (1..5)", "check_row", {}),
    ("reviews", "UNIQUE (customer_id, product_id)", "unique_chosen", {}),
    ("audit", "at NOT NULL", "not_null", {}),
]
RULE = {
    "not_null": (True, "row-local: any merge of valid rows is valid"),
    "check_row": (True, "row-local: any merge of valid rows is valid"),
    "unique_generated": (True, "uniqueness by construction (uuid or a per-replica id range)"),
    "fk_insert_only": (True, "inserts only: a referenced row never disappears"),
    "unique_chosen": (False, "two replicas can accept the same chosen value: needs a unique index on one node, or a reservation"),
    "fk_with_deletes": (False, "a delete on one replica and an insert on another can orphan a row: needs coordination on delete"),
    "lower_bound": (False, "two replicas can each spend the same slack: needs a lock, or escrow (pre-split slack)"),
    "upper_bound": (False, "two replicas can each grant the last unit: needs a lock, or escrow"),
}
rows = []
for table, c, kind, d in SCHEMA:
    ok, why = RULE[kind]
    r = {"table": table, "constraint": c, "i_confluent": "yes" if ok else "no", "why": why}
    if "slack" in d:                                  # escrow: each replica gets slack / replicas, refilled when spent
        per = d["slack"] / args.replicas
        r["escrow_refills_per_hour"] = d["use_per_s"] / per * 3600
    rows.append(r)
    print(f"{table:<12} {c:<62} {'I-confluent' if ok else 'NEEDS COORDINATION'}"
          + (f"   escrow: {r['escrow_refills_per_hour']:.1f} refills/hour" if "escrow_refills_per_hour" in r else ""))
need = sum(r["i_confluent"] == "no" for r in rows)
print(f"\n{need} of {len(rows)} constraints ({need / len(rows):.0%}) need coordination")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"constraints": rows, "summary": [{"need": need, "total": len(rows),
                   "need_pct": need / len(rows) * 100}]}}, f)