"""An in-process pub/sub broker with topics, fan-out, and bounded queues.

One publish lands in every subscriber's own queue, so a subscriber that
consumes slowly falls behind on its own without holding anybody else up.
That isolation is the reason per-subscriber queues exist — and it is also
what forces the question this toy is about, because a queue that is private
is also a queue that is *bounded*, and a bounded queue eventually has to
refuse something.

When a slow subscriber's queue is full and another message arrives, exactly
one message is not going to be delivered. Which one is the overflow policy:

  drop-newest  refuse the arrival; the queue keeps what it already had
  drop-oldest  evict the head to make room; the queue keeps the freshest
  block        refuse the *publish*, for every subscriber on the topic

There is no fourth option that keeps everything, and the first two do not
differ in how much survives. See the commentary.

Time is an integer tick supplied by the caller. Nothing sleeps, no thread is
started, and there is no randomness anywhere — the same program prints the
same numbers on every machine.
"""

from __future__ import annotations

from collections import deque

DROP_NEWEST = "drop-newest"
DROP_OLDEST = "drop-oldest"
BLOCK = "block"


class Subscriber:
    """One consumer, with a private bounded queue and a drain rate.

    `drain_rate` is how many messages it manages to consume per tick — this
    toy's entire model of "slow consumer". A real one is slow because of what
    it does with each message; the arithmetic of falling behind is the same.
    """

    def __init__(self, name: str, drain_rate: int = 1, capacity: int = 10,
                 policy: str = DROP_NEWEST):
        if policy not in (DROP_NEWEST, DROP_OLDEST, BLOCK):
            raise ValueError(policy)
        self.name = name
        self.drain_rate = drain_rate
        self.capacity = capacity
        self.policy = policy
        self.queue: deque[tuple[int, int]] = deque()  # (seq, published_at)
        self.delivered: list[tuple[int, int, int]] = []  # (seq, published, received)
        self.dropped = 0

    @property
    def full(self) -> bool:
        return len(self.queue) >= self.capacity

    def offer(self, seq: int, published_at: int) -> bool:
        """Try to enqueue one message. Returns whether it was accepted.

        Both drop policies leave the queue at exactly `capacity` afterwards.
        That is not a coincidence and it is the whole result: the occupancy
        trajectory does not depend on which message you throw away, so the
        *count* of thrown-away messages cannot depend on it either.
        """
        if not self.full:
            self.queue.append((seq, published_at))
            return True
        if self.policy == DROP_NEWEST:
            self.dropped += 1
            return False
        if self.policy == DROP_OLDEST:
            self.queue.popleft()
            self.queue.append((seq, published_at))
            self.dropped += 1
            return False
        raise AssertionError("a BLOCK subscriber is never offered a message "
                             "it has no room for; the broker checks first")

    def drain(self, now: int) -> int:
        """Consume up to `drain_rate` messages. Returns how many."""
        taken = 0
        while taken < self.drain_rate and self.queue:
            seq, published_at = self.queue.popleft()
            self.delivered.append((seq, published_at, now))
            taken += 1
        return taken

    def ages(self) -> list[int]:
        """Ticks between publication and delivery, per delivered message."""
        return [received - published for _, published, received in self.delivered]


class Broker:
    def __init__(self):
        self.topics: dict[str, list[Subscriber]] = {}
        self.published = 0
        self.refused = 0  # publishes rejected outright by a BLOCK subscriber

    def subscribe(self, topic: str, subscriber: Subscriber) -> Subscriber:
        self.topics.setdefault(topic, []).append(subscriber)
        return subscriber

    def publish(self, topic: str, now: int) -> bool:
        """Fan one message out to every subscriber on the topic.

        Backpressure is a property of the *topic*, not of one consumer: if a
        blocking subscriber has no room, nobody gets the message, because the
        alternative is to deliver it to some subscribers and not others and
        call that the same stream.
        """
        subscribers = self.topics.get(topic, [])
        if any(s.policy == BLOCK and s.full for s in subscribers):
            self.refused += 1
            return False
        seq = self.published
        self.published += 1
        for s in subscribers:
            s.offer(seq, now)
        return True

    def tick(self, now: int) -> None:
        for subscribers in self.topics.values():
            for s in subscribers:
                s.drain(now)


def run(policy: str, ticks: int = 200, publish_rate: int = 5, capacity: int = 10,
        drain_rates: tuple[int, ...] = (1, 5)) -> tuple[Broker, list[Subscriber]]:
    """One standard run: N subscribers on one topic, drained then published.

    Draining before publishing within a tick is a choice, not a law — it
    means a subscriber's capacity is measured *after* it has done this tick's
    work. Publishing first would shift every number by one queue slot without
    changing anything the toy is about.
    """
    broker = Broker()
    subs = [
        broker.subscribe("events", Subscriber(f"sub{i}", drain_rate=r,
                                              capacity=capacity, policy=policy))
        for i, r in enumerate(drain_rates)
    ]
    for now in range(ticks):
        broker.tick(now)
        for _ in range(publish_rate):
            broker.publish("events", now)
    return broker, subs