#!/usr/bin/env python3
"""Losing causal order across two logs notifies the ex you unfriended, DDIA 2e Ch. 13.

The book's break-up example. In the real world Alice does two things, in order:
first she UNFRIENDS her ex Bob, then she POSTS a rude message. Because she
unfriended him *before* posting, Bob must not be notified about the post.

But the two facts live in two INDEPENDENT logs -- a friendship log and a message
log -- with no shared order between them. A notification service is a join that
consumes both. Merge them by ARRIVAL order and, because the logs travel over
independent transports, the post can arrive BEFORE the unfriend. The consumer
then still sees Alice and Bob as friends and FIRES a notification to Bob: a real
side effect that cannot be un-sent.

Two fixes the chapter names close the leak:
  (a) route both events for the (alice, bob) pair through the SAME ordered
      partition, so the unfriend is delivered before the post; or
  (b) stamp the post with a CAUSAL REFERENCE to the friend-list version it was
      written against, and have the consumer WAIT for that version before
      processing the post.

Pure standard library. The log arrival order is modeled explicitly, so the run
is deterministic and reproducible.
"""

# The real-world causal order Alice performed, and can never change:
#   1. unfriend Bob   -> Alice's friend-list version goes 1 -> 2
#   2. post the message (written while NOT friends with Bob)
#
# The friend-list VERSION is the hinge. Version 1 = still friends with Bob;
# version 2 = Bob removed. The post was authored against version 2.

FRIEND_VERSION_BEFORE = 1   # Alice & Bob are friends
FRIEND_VERSION_AFTER = 2    # Alice has unfriended Bob

# --- The two independent logs -----------------------------------------------

# Friendship log: Alice unfriends Bob. Applying it removes the friendship and
# advances Alice's friend-list version to 2.
unfriend_event = {
    "log": "friendship",
    "type": "unfriend",
    "user": "alice",
    "target": "bob",
    "new_friend_version": FRIEND_VERSION_AFTER,
}

# Message log: Alice posts a rude message. It was written AFTER the unfriend, so
# in reality it depends on friend-list version 2. Fix (b) records that; the
# buggy consumer and fix (a) do not need the field.
post_event = {
    "log": "message",
    "type": "post",
    "author": "alice",
    "text": "good riddance",
    "depends_on_friend_version": FRIEND_VERSION_AFTER,
}


def fresh_world():
    """The notification service's starting view: Alice and Bob are still friends,
    and Alice's friend list is at version 1."""
    return {
        "friends_of_alice": {"bob"},   # who would be notified about Alice's posts
        "alice_friend_version": FRIEND_VERSION_BEFORE,
    }


def apply_unfriend(world):
    world["friends_of_alice"].discard("bob")
    world["alice_friend_version"] = unfriend_event["new_friend_version"]


def notify_for_post(world):
    """Return the set of users notified about Alice's post: her current friends."""
    return set(world["friends_of_alice"])


# --- Buggy consumer: join two logs, merge by ARRIVAL order ------------------

def run_buggy():
    """The logs travel over independent transports, so the post arrives before
    the unfriend. Merging by arrival order, the consumer processes the post while
    Alice and Bob still look like friends -> Bob is notified."""
    world = fresh_world()
    # Independent transport: the message log happens to arrive first.
    arrival_order = [post_event, unfriend_event]

    notified = set()
    for event in arrival_order:
        if event["type"] == "post":
            hit = notify_for_post(world)
            notified |= hit
        elif event["type"] == "unfriend":
            apply_unfriend(world)
    return notified


# --- Fix (a): one ordered partition for the (alice, bob) pair ---------------

def run_fix_single_partition():
    """Route BOTH events for the (alice, bob) pair through the same partition.
    A single log has one order, and we place the events in causal order:
    unfriend, then post. The consumer reads them in that order."""
    world = fresh_world()
    # One log, causal order preserved: unfriend delivered before post.
    partition = [unfriend_event, post_event]

    notified = set()
    for event in partition:
        if event["type"] == "unfriend":
            apply_unfriend(world)
        elif event["type"] == "post":
            notified |= notify_for_post(world)
    return notified


# --- Fix (b): causal reference the consumer waits on ------------------------

def run_fix_causal_dependency():
    """Keep two independent logs and the SAME adversarial arrival order (post
    first). But the post carries depends_on_friend_version = 2. The consumer
    buffers any post whose required friend-list version it has not yet applied,
    and only processes it once that version arrives."""
    world = fresh_world()
    arrival_order = [post_event, unfriend_event]

    notified = set()
    buffer = []

    def try_process(post):
        # Only safe to process once the friend-list version the post was written
        # against has been applied.
        if world["alice_friend_version"] >= post["depends_on_friend_version"]:
            notified.update(notify_for_post(world))
            return True
        return False

    for event in arrival_order:
        if event["type"] == "post":
            if not try_process(event):
                buffer.append(event)          # dependency unmet -> wait
        elif event["type"] == "unfriend":
            apply_unfriend(world)
            # A version advanced; retry anything that was waiting on it.
            buffer[:] = [p for p in buffer if not try_process(p)]

    return notified


def report(label, notified):
    leaked = "bob" in notified
    verdict = "YES -- LEAK" if leaked else "NO  -- suppressed"
    who = ", ".join(sorted(notified)) if notified else "(no one)"
    print(f"  notification sent to ex (bob)?  {verdict}")
    print(f"  users notified about the post:  {who}")
    return leaked


def main():
    print("Alice's real-world causal order:  1) unfriend bob   2) post \"good riddance\"")
    print("Because she unfriended bob BEFORE posting, bob must NOT be notified.")
    print("The two facts live in two independent logs with no shared order.\n")

    print("[buggy] one consumer joins both logs, merged by ARRIVAL order")
    print("        independent transport delivers the post before the unfriend:")
    report("buggy", run_buggy())
    print()

    print("[fix a] route both events for the (alice, bob) pair through ONE ordered")
    print("        partition -> unfriend is delivered before the post:")
    report("fix-a", run_fix_single_partition())
    print()

    print("[fix b] two logs, SAME arrival order (post first), but the post carries")
    print("        depends_on_friend_version=2; the consumer waits for that version:")
    report("fix-b", run_fix_causal_dependency())


if __name__ == "__main__":
    main()