# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Replay determinism test.

A transformation that labels orders as overdue using current_timestamp, run
twice a little apart, and run with the clock moved a day (as a rerun tomorrow
would); then the same transformation with an explicit run_date parameter.
Compares the outputs of each pair by hash.
"""
import argparse, hashlib, json, sqlite3, time

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--orders", type=int, default=10_000, help="Orders.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

c = sqlite3.connect(":memory:")
c.execute("CREATE TABLE orders (id INT, due TEXT)")
base = time.time()
c.executemany("INSERT INTO orders VALUES (?, datetime(?, 'unixepoch'))", ((i, base - 86400 + i * 17) for i in range(args.orders)))   # due dates around now
def ambient(shift_days=0):
    q = f"SELECT id, due < datetime('now', '{shift_days:+d} days') AS overdue FROM orders ORDER BY id"
    return hashlib.sha256(str(c.execute(q).fetchall()).encode()).hexdigest()[:16]
def parameterized(run_date):
    q = "SELECT id, due < ? AS overdue FROM orders ORDER BY id"
    return hashlib.sha256(str(c.execute(q, (run_date,)).fetchall()).encode()).hexdigest()[:16]
run_date = c.execute("SELECT datetime(?, 'unixepoch')", (base,)).fetchone()[0]
first = ambient(); time.sleep(2); second = ambient(); tomorrow = ambient(1)
p1 = parameterized(run_date); p2 = parameterized(run_date)
rows = [{"version": "current_timestamp", "run": "now", "hash": first}, {"version": "current_timestamp", "run": "2 s later", "hash": second},
        {"version": "current_timestamp", "run": "rerun tomorrow", "hash": tomorrow},
        {"version": "run_date parameter", "run": "today", "hash": p1}, {"version": "run_date parameter", "run": "rerun tomorrow, same run_date", "hash": p2}]
summary = {"ambient_identical": "yes" if first == second == tomorrow else "no", "param_identical": "yes" if p1 == p2 else "no"}
for r in rows: print(f"{r['version']:<20} {r['run']:<32} {r['hash']}")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"summary": summary, "tables": {"runs": rows}}, f)