"""What a Snowflake ID is made of, and what actually keeps it unique.

    python3 demo.py

One trace throughout: machine 7, 8 IDs/ms, 200 ms of true time, and one 50 ms
NTP step backwards at true t=100 ms. No sleeping and no wall clock is ever
read: `Wall` derives the clock from the simulation's own counter, so the whole
thing runs instantly and prints identical output on every machine. The single
RNG (section 7 only) is seeded.
"""

import datetime as dt
import random

from snowflake import (BLIND, BORROW, CLAMP, RAM, REFUSE, RESUME, SKIP,
                       Exhausted, Snowflake, run)

RATE, DUR, STEP_AT, STEP_MS = 8, 200, 100.0, 50
OFFERED = RATE * DUR


def rule(title):
    print(f"\n{title}")
    print("-" * len(title))


print(f"machine 7, {RATE} IDs/ms for {DUR} ms = {OFFERED} requests offered.")
print(f"At true t={STEP_AT:.0f} ms the wall clock steps back {STEP_MS} ms.")

# ---------------------------------------------------------------- 1
rule("1. The 64 bits")

g = Snowflake(machine_id=7, rewind=CLAMP)
ident, _ = g.next_id(50)
ts, mid, seq = g.unpack(ident)
print(f"next_id(50) -> {ident}")
print(f"   {ident:064b}")
print(f"   unpacks to ts={ts} machine={mid} seq={seq}")
print(f"   fields: {g.ts_bits} ts bits, {g.mid_bits} machine bits "
      f"({g.mid_max + 1} machines), {g.seq_bits} seq bits "
      f"({g.seq_mask + 1} IDs per ms per machine)")

span = 1 << 41
twitter_epoch_ms = 1288834974657
born = dt.datetime.fromtimestamp(twitter_epoch_ms / 1000, dt.UTC)
dies = dt.datetime.fromtimestamp((twitter_epoch_ms + span) / 1000, dt.UTC)
print(f"\n41 bits of ms = {span} ms = {span / 1000 / 86400 / 365.2425:.2f} years")
print(f"   epoch {born:%Y-%m-%d} -> the timestamp field overflows {dies:%Y-%m-%d}")

a, b = Snowflake(machine_id=1), Snowflake(machine_id=1025)
ia, _ = a.next_id(500)
ib, _ = b.next_id(500)
print(f"\nmachine_id=1    -> {ia}")
print(f"machine_id=1025 -> {ib}   identical: {ia == ib}  (1025 & 1023 = {1025 & 1023})")
print("The field is masked, not checked. Machine 1025 is machine 1.")

# ---------------------------------------------------------------- 2
rule("2. The clock steps back 50 ms")

print(f"{'rewind policy':>14} {'issued':>8} {'refused':>8} {'DUPES':>7} {'stall(ms)':>10}")
base = {}
for policy in (BLIND, REFUSE, CLAMP):
    r = run(policy, rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS)
    base[policy] = r
    print(f"{policy:>14} {r.issued:>8} {r.refused:>8} {r.dupes:>7} {r.stall:>10.3f}")

collisions = base[BLIND].collisions()
print(f"\nfirst 3 collisions under {BLIND}, as (t_first, t_again, id):")
for first, again, i in collisions[:3]:
    ts, mid, seq = base[BLIND].gen.unpack(i)
    print(f"   t={first:7.3f} and t={again:7.3f} -> {i}  (ts={ts} machine={mid} seq={seq})")
print(f"total {len(collisions)} = rate {RATE}/ms x rewind {STEP_MS} ms "
      f"= {RATE * STEP_MS}")
print(f"\n{REFUSE} pays {base[REFUSE].refused} refusals for its 0 duplicates; "
      f"{CLAMP} pays nothing.")

# ---------------------------------------------------------------- 3
rule("3. Now add a liveness probe: K consecutive refusals and it restarts")

ks = (0, 1, 3, 10, 50)
print(f"{'policy':>8} " + " ".join(f"{'K=' + str(k):>7}" for k in ks) + "   restarts")
for policy in (BLIND, REFUSE, CLAMP):
    rs = [run(policy, rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS,
              fail_after=k) for k in ks]
    print(f"{policy:>8} " + " ".join(f"{r.dupes:>7}" for r in rs)
          + "   " + " ".join(str(r.restarts) for r in rs))

r10 = run(REFUSE, rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS,
          fail_after=10)
t_restart, w_restart = r10.restart_log[0]
print(f"\nThe defence that refuses to serve produced {r10.dupes} duplicates.")
print(f"Doing nothing at all produced {base[BLIND].dupes}. It bought "
      f"{base[BLIND].dupes - r10.dupes}.")
print(f"\nderivation, K=10: refusals start the instant the clock steps, at true")
print(f"   t={STEP_AT:.3f}, one every 1/{RATE} ms. The 10th is 9 intervals later:")
print(f"   {STEP_AT:.3f} + 9/{RATE} = {STEP_AT + 9 / RATE:.3f}  "
      f"(measured restart: t={t_restart:.3f}, wall read {w_restart})")
print(f"   the next request arrives at t={t_restart + 1 / RATE:.3f}, and every one")
print(f"   from there until the wall recovers at t={STEP_AT + STEP_MS:.3f} is a duplicate:")
print(f"   ({STEP_AT + STEP_MS:.3f} - {t_restart + 1 / RATE:.3f}) x {RATE} = "
      f"{(STEP_AT + STEP_MS - t_restart - 1 / RATE) * RATE:.0f}")
print(f"\nclosed form, dupes = {base[BLIND].dupes} - K:")
for k in ks[1:]:
    got = run(REFUSE, rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS,
              fail_after=k).dupes
    print(f"   K={k:<3} {base[BLIND].dupes} - {k:<3} = {base[BLIND].dupes - k:<3} "
          f"measured {got}")

# ---------------------------------------------------------------- 4
rule("4. The amnesia is one integer. Persisting it is not enough.")

RESTART_AT = 120.0
rules = ((RAM, "last_ts = -1        "), (RESUME, "last_ts = hwm       "),
         (SKIP, "last_ts = hwm + 1   "))
print(f"restart at true t={RESTART_AT:.0f} ms, 20 ms into the rewound window\n")
print(f"{'startup rule':>22} {'blind':>8} {'refuse':>8} {'clamp':>8}")
for start_rule, label in rules:
    row = [run(p, rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS,
               restart_at=RESTART_AT, start_rule=start_rule).dupes
           for p in (BLIND, REFUSE, CLAMP)]
    print(f"{label:>22} " + " ".join(f"{d:>8}" for d in row))

resume_clamp = run(CLAMP, rate=RATE, duration=DUR, step_at=STEP_AT,
                   step_ms=STEP_MS, restart_at=RESTART_AT, start_rule=RESUME)
spent = RATE + int(RESTART_AT - STEP_AT) * RATE
print(f"\nwhy `last_ts = hwm` still collides, derived: the high-water mark at the")
print(f"restart is ms {int(STEP_AT) - 1}. The old generator spent that millisecond as")
print(f"   seq 0..{RATE - 1}     its {RATE} genuine requests during true t={STEP_AT - 1:.0f}..{STEP_AT:.0f}")
print(f"   seq {RATE}..{spent - 1}   {int(RESTART_AT - STEP_AT)} ms of clamped requests, {RATE}/ms")
print(f"   {RATE} + {int(RESTART_AT - STEP_AT)}x{RATE} = {spent} slots, so seq 0..{spent - 1} were used.")
print(f"The replacement resumes AT ms {int(STEP_AT) - 1} with seq=0 and reissues seq 1..{spent - 1}:")
print(f"   {spent - 1} duplicates. Measured: {resume_clamp.dupes}.")
print("The millisecond is not the unit of uniqueness; the (ms, seq) pair is.")

# ---------------------------------------------------------------- 5
rule("5. Two one-line edits")


class Leq(Snowflake):
    """`if now < self.last_ts` in next_id, with the `<` changed to `<=`."""

    def next_id(self, now):
        if now <= self.last_ts:
            if self.rewind == REFUSE:
                self.refusals += 1
                raise Exhausted("clock not ahead of last_ts")
            if self.rewind == CLAMP:
                now = self.last_ts
        return Snowflake.next_id(self, now)


class FreeRunning(Snowflake):
    """`self.seq = 0` on a new millisecond, deleted. The counter never resets."""

    def next_id(self, now):
        if now < self.last_ts and self.rewind == CLAMP:
            now = self.last_ts
        if now == self.last_ts:
            self.seq = (self.seq + 1) & self.seq_mask
            if self.seq == 0:
                return self._overflow(now)
        else:
            self.last_ts = now
            self.seq = (self.seq + 1) & self.seq_mask
        return self.pack(self.last_ts, self.seq), self.last_ts


print("(a) `now < last_ts` -> `now <= last_ts`, no restart\n")
print(f"{'generator':>16} {'policy':>8} {'issued':>8} {'refused':>8} {'DUPES':>7}")
for cls, name in ((Snowflake, "<  as written"), (Leq, "<= variant")):
    for policy in (REFUSE, CLAMP):
        r = run(policy, rate=RATE, duration=DUR, step_at=STEP_AT,
                step_ms=STEP_MS, cls=cls)
        print(f"{name:>16} {policy:>8} {r.issued:>8} {r.refused:>8} {r.dupes:>7}")
print("\ninert for clamp; for refuse it now rejects every request that shares a")
print(f"millisecond with the previous one, i.e. {RATE - 1} of every {RATE}.")

print("\n(b) drop `self.seq = 0` on a new millisecond, no restart\n")
print(f"{'generator':>16} {'policy':>8} {'issued':>8} {'refused':>8} {'DUPES':>7}")
for cls, name in ((Snowflake, "seq resets"), (FreeRunning, "free-running seq")):
    r = run(BLIND, rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS,
            cls=cls)
    print(f"{name:>16} {BLIND:>8} {r.issued:>8} {r.refused:>8} {r.dupes:>7}")
print("\nA free-running counter survives the rewind with no clock defence at all.")
print("It is not safe, though, it is phase-shifted: the replacement's counter")
print("starts at 0 while the old one was at rate x t mod 4096, and they collide")
print(f"whenever a restart sits a whole number of 4096/{RATE} = "
      f"{4096 // RATE} ms after the step.")
LONG = dict(rate=RATE, duration=1000, step_at=600.0, step_ms=200)
print(f"\nlonger trace: {RATE} IDs/ms, 200 ms rewind at t=600, restart at t=R\n")
print(f"{'R':>7} {'R - rewind':>11} {'DUPES':>7}")
for at in (700.0, 704.0, 708.0, 712.0, 716.0, 720.0):
    r = run(BLIND, cls=FreeRunning, restart_at=at, **LONG)
    print(f"{at:>7.0f} {at - 200:>11.0f} {r.dupes:>7}")
print(f"\n712 - 200 = 512 exactly, and the collision is the whole rest of the")
print(f"window: wall ms 512..599, {RATE} IDs each = {(800 - 712) * RATE}.")
print(f"The ordinary generator on that same trace: "
      f"{run(BLIND, **LONG).dupes} duplicates.")

# ---------------------------------------------------------------- 6
rule("6. The boundary: clamping spends the sequence field, so rewind has a budget")

print(f"50 ms rewind, sweeping the request rate (12 seq bits = 4096 slots/ms)\n")
print(f"{'IDs/ms':>7} {'issued':>9} {'DUPES':>7} {'stall(ms)':>11} {'max stall':>11}")
for rate in (8, 40, 79, 80, 81, 82, 100, 400):
    r = run(CLAMP, rate=rate, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS)
    print(f"{rate:>7} {r.issued:>9} {r.dupes:>7} {r.stall:>11.3f} {r.max_stall:>11.3f}")
print(f"\nslots needed = rate x (rewind + 1) ; available = 4096")
print(f"   51 x 80 = {51 * 80} <= 4096      51 x 81 = {51 * 81} > 4096")

print("\nheld from the other side: 100 IDs/ms, sweeping the rewind\n")
print(f"{'rewind ms':>10} {'DUPES':>7} {'stall(ms)':>11}")
for step in (1, 10, 39, 40, 41, 50):
    r = run(CLAMP, rate=100, duration=DUR, step_at=STEP_AT, step_ms=step)
    print(f"{step:>10} {r.dupes:>7} {r.stall:>11.3f}")
print(f"   100 x 40 = {100 * 40} <= 4096      100 x 41 = {100 * 41} > 4096")

print("\nso the bit split is not machines-vs-throughput. It is machines vs. how")
print("much clock error you can absorb. Same 50 ms rewind at 100 IDs/ms:\n")
print(f"{'layout':>12} {'machines':>9} {'seq slots':>10} {'stall(ms)':>11} {'DUPES':>7}")
for mid_bits, seq_bits in ((11, 11), (10, 12), (9, 13), (8, 14)):
    r = run(CLAMP, rate=100, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS,
            mid_bits=mid_bits, seq_bits=seq_bits)
    print(f"{f'41/{mid_bits}/{seq_bits}':>12} {1 << mid_bits:>9} {1 << seq_bits:>10} "
          f"{r.stall:>11.3f} {r.dupes:>7}")
print("\nOne bit moved off the machine id takes the stall to zero and halves the")
print("fleet cap. `borrow` refuses to stall at all, and pays in a different coin:")
r = run(CLAMP, BORROW, rate=100, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS)
ahead = max(r.gen.unpack(i)[0] - w for _t, w, i, o in r.records if o == "ok")
print(f"   clamp/borrow at 100 IDs/ms: {r.dupes} dupes, 0 stall, "
      f"{r.borrowed} borrowed ms,")
print(f"   and timestamps running up to {ahead} ms ahead of the wall clock.")

print("\nAnd where the whole problem vanishes: a rewind of one millisecond.\n")
print(f"{'rewind ms':>10} {'blind DUPES':>12} {'refuse refusals':>16}")
for step in (0, 1, 2, 3):
    small = dict(rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=step)
    print(f"{step:>10} {run(BLIND, **small).dupes:>12} "
          f"{run(REFUSE, **small).refused:>16}")
print(f"\ndupes = rate x rewind for a rewind of 2 ms or more, and zero at 1 ms:")
print("a step back that lands on the millisecond the generator is already in")
print("takes the `now == last_ts` branch and just keeps counting. Every policy")
print("is inert there, including having none.")

# ---------------------------------------------------------------- 7
rule("7. The claim that survives: roughly sortable, and 'roughly' means 1 ms")


def inversions(seq):
    """Pairs out of order, counted by merge sort."""
    if len(seq) < 2:
        return seq, 0
    mid = len(seq) // 2
    left, a = inversions(seq[:mid])
    right, b = inversions(seq[mid:])
    merged, i, j, n = [], 0, 0, a + b
    while i < len(left) and j < len(right):
        if left[i] <= right[j]:
            merged.append(left[i])
            i += 1
        else:
            merged.append(right[j])
            n += len(left) - i
            j += 1
    return merged + left[i:] + right[j:], n


print(f"{'machines':>9} {'IDs/ms':>7} {'IDs':>7} {'pairs':>10} "
      f"{'mis-ordered':>17} {'worst error':>13}")
for machines, per_ms, ms in ((1, 10, 200), (8, 1, 200), (8, 100, 20), (64, 10, 50)):
    rng = random.Random(20260803)
    gens = [Snowflake(machine_id=m, rewind=CLAMP, overflow=BORROW)
            for m in range(machines)]
    events = []
    for msec in range(ms):
        for m in range(machines):
            for off in sorted(rng.random() for _ in range(per_ms)):
                events.append((msec + off, gens[m].next_id(msec)[0]))
    events.sort()                       # true chronological order
    rank = {i: r for r, (_t, i) in enumerate(events)}
    order = [rank[i] for _t, i in sorted(events, key=lambda e: e[1])]
    _, inv = inversions(order)
    worst, high = 0.0, -1.0
    for r in order:                     # id order; how far back does time jump?
        high = max(high, events[r][0])
        worst = max(worst, high - events[r][0])
    pairs = len(order) * (len(order) - 1) // 2
    print(f"{machines:>9} {per_ms:>7} {len(order):>7} {pairs:>10} "
          f"{inv:>10} {100 * inv / pairs:>5.1f}% {worst:>10.3f} ms")
print("\nTrue, and boring: every inversion is under 1 ms because sub-millisecond")
print("order is not in the ID at all. One machine cannot invert anything.")