#!/usr/bin/env python3
"""DDIA Ch. 12 - Stream Processing: time-dependence of a stream-table join.

A stream of historical SALES is joined against a TAX-RATE table that changed
over time (8% before day 100, 10% from day 100 on). A naive stream-table join
looks up the table's CURRENT state at processing time, so reprocessing last
year's sales applies this year's rate -> wrong invoice totals, and rerunning
the job on a different day gives a DIFFERENT answer (non-deterministic replay).

A point-in-time / temporal join instead matches each sale to the rate that was
in effect AT SALE TIME (a slowly-changing dimension looked up by effective
date), so the totals are correct AND identical no matter when the job runs.

Everything is deterministic: no wall clock, no randomness. The "when did the
job run" is passed in explicitly as a run-day, so the two runs reproduce
exactly. Money is tracked in integer cents; rates in basis points; the join is
therefore exact.
"""

# --- The tax-rate table: a slowly-changing dimension with effective dates ---
# Each version is (effective_from_day, rate_in_basis_points). Sorted ascending.
# 800 bp = 8.00%, 1000 bp = 10.00%. The rate changed on day 100.
TAX_TABLE = [
    (0,   800),    # from day 0:   8%
    (100, 1000),   # from day 100: 10%
]

# --- The sales stream: (sale_id, day, amount_cents). Two before the day-100
# rate change (should be taxed at 8%), two after (should be taxed at 10%). ---
SALES = [
    ("S1", 40,  10000),   # $100.00 on day 40   -> should be 8%
    ("S2", 80,  20000),   # $200.00 on day 80   -> should be 8%
    ("S3", 120, 30000),   # $300.00 on day 120  -> should be 10%
    ("S4", 160, 40000),   # $400.00 on day 160  -> should be 10%
]


def rate_as_of(day):
    """Point-in-time lookup: the tax rate (bp) in effect on `day`.

    Returns the latest table version whose effective_from is <= day. This is
    the temporal / SCD lookup -- it depends only on the argument, never on when
    the function is called.
    """
    rate = TAX_TABLE[0][1]
    for eff_from, bp in TAX_TABLE:
        if eff_from <= day:
            rate = bp
    return rate


def current_rate(run_day):
    """Current-state lookup: the table's LATEST version as of the wall clock.

    A naive stream-table join reads the table row live at processing time, so
    the answer it gets is whatever version is current WHEN THE JOB RUNS
    (run_day) -- not when the sale happened. This is the hidden input.
    """
    return rate_as_of(run_day)


def tax_cents(amount_cents, bp):
    # Exact integer arithmetic; all products here divide evenly.
    return amount_cents * bp // 10000


def dollars(cents):
    return f"${cents / 100:,.2f}"


def pct(bp):
    return f"{bp / 100:g}%"


def naive_join(run_day):
    """Join every sale against the table's CURRENT state at run_day."""
    bp = current_rate(run_day)
    total = 0
    rows = []
    for sid, day, amt in SALES:
        t = tax_cents(amt, bp)
        total += t
        rows.append((sid, day, amt, bp, t))
    return bp, total, rows


def pointintime_join(run_day):
    """Join every sale against the rate effective AT THE SALE'S OWN day.

    run_day is accepted only to prove it is ignored -- the result is a pure
    function of the sales + the table history.
    """
    total = 0
    rows = []
    for sid, day, amt in SALES:
        bp = rate_as_of(day)
        t = tax_cents(amt, bp)
        total += t
        rows.append((sid, day, amt, bp, t))
    return total, rows


def print_rows(rows):
    for sid, day, amt, bp, t in rows:
        print(f"      {sid}  day {day:>3}  amount {dollars(amt):>9}"
              f"  x {pct(bp):>4}  ->  tax {dollars(t):>7}")


RULE = "=" * 66

# Two runs of the SAME job on the SAME sales, only the calendar day differs.
RUN_EARLY = 90    # job run before the rate change: current rate is still 8%
RUN_LATE = 300    # job re-run after the rate change: current rate is now 10%

sales_total = sum(a for _, _, a in SALES)


def main():
    print(RULE)
    print("Stream-table join: historical SALES  x  a TAX-RATE table that changed")
    print(RULE)
    print()
    print("Tax-rate table (a slowly-changing dimension, effective-dated):")
    for eff_from, bp in TAX_TABLE:
        print(f"    from day {eff_from:>3}:  {pct(bp)}")
    print()
    print("Sales stream (total sales = " + dollars(sales_total) + "):")
    for sid, day, amt in SALES:
        print(f"    {sid}  day {day:>3}  amount {dollars(amt):>9}")
    print()

    # --- Step 1: naive join, re-run LATE (today). Current rate = 10%. ---
    print(RULE)
    print(f"Step 1  NAIVE join, job run on day {RUN_LATE} (today).")
    print("        Looks up the table's CURRENT rate for EVERY sale.")
    print(RULE)
    bp_late, naive_late, rows_late = naive_join(RUN_LATE)
    print(f"  current rate on day {RUN_LATE} = {pct(bp_late)}  (applied to all sales)")
    print_rows(rows_late)
    print(f"  NAIVE total tax = {dollars(naive_late)}")
    print()

    # --- Step 2: same naive job, re-run EARLY (last year). Current rate = 8%. ---
    print(RULE)
    print(f"Step 2  SAME naive job, re-run on day {RUN_EARLY} (last year).")
    print("        Same sales, same code -- only the calendar day differs.")
    print(RULE)
    bp_early, naive_early, rows_early = naive_join(RUN_EARLY)
    print(f"  current rate on day {RUN_EARLY} = {pct(bp_early)}  (applied to all sales)")
    print_rows(rows_early)
    print(f"  NAIVE total tax = {dollars(naive_early)}")
    print()
    drift = naive_late - naive_early
    print(f"  >> Same input, two run days, two answers: "
          f"{dollars(naive_early)} vs {dollars(naive_late)}")
    print(f"  >> Reprocessing DRIFT = {dollars(drift)} "
          f"-- the join is NOT deterministic on replay.")
    print()

    # --- Step 3: point-in-time join, both run days. ---
    print(RULE)
    print("Step 3  POINT-IN-TIME join: each sale x the rate effective at ITS day.")
    print(RULE)
    pit_late, pit_rows = pointintime_join(RUN_LATE)
    print_rows(pit_rows)
    print(f"  CORRECT total tax = {dollars(pit_late)}")
    print()
    pit_early, _ = pointintime_join(RUN_EARLY)
    print(f"  re-run on day {RUN_EARLY}  -> total tax = {dollars(pit_early)}")
    print(f"  re-run on day {RUN_LATE}  -> total tax = {dollars(pit_late)}")
    stable = "IDENTICAL" if pit_early == pit_late else "DIFFERENT"
    print(f"  >> Both runs {stable}: {dollars(pit_early)} "
          f"-- deterministic on replay.")
    print()

    # --- Summary ---
    print(RULE)
    print("Summary")
    print(RULE)
    print(f"  naive, run today (day {RUN_LATE}, current rate {pct(bp_late)}):  "
          f"{dollars(naive_late)}   (WRONG: old sales overcharged)")
    print(f"  naive, re-run last year (day {RUN_EARLY}, current rate {pct(bp_early)}): "
          f"{dollars(naive_early)}   (WRONG, and DIFFERENT: drift {dollars(drift)})")
    print(f"  point-in-time (either run day):              "
          f"{dollars(pit_late)}   (CORRECT and STABLE)")


if __name__ == "__main__":
    main()