"""Идемпотентность: «ровно один раз» бывает в эффекте, а не в доставке.

ЗАЧЕМ ЭТОТ ФАЙЛ. «Сделаем exactly-once» — обещание, которое дают чаще, чем
выполняют. Проверить его словами нельзя: нужен сервер, клиент, повтор и
счётчик того, сколько раз эффект случился на самом деле.

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ. Всё на loopback, оба конца в одном процессе:
  1. Повтор без ключа: сколько раз применился эффект.
  2. Повтор с ключом идемпотентности: сколько раз применился эффект.
  3. Окно между записью эффекта и отправкой ответа: что видит клиент,
     если ответ потерялся, и что при этом уже произошло на сервере.
  4. Одновременные повторы: два запроса с одним ключом в один момент.
  5. Ключ, выданный клиентом, против ключа, выданного сервером.

ЧЕГО ЗДЕСЬ НЕТ. Ни сети между машинами, ни падений сервера, ни хранилища с
транзакциями: «база» — это словарь в памяти. Предмет замера — число
применений эффекта, а не долговечность записи.

ПОЧЕМУ ПОДПИСИ ПО-АНГЛИЙСКИ. Урок существует в двух языках и цитирует запись
прогона дословно обеими версиями.

ЗАПУСК: python3 bench/idempotency/keys.py
Вывод: runs/keys.txt
"""

import os
import socket
import sys
import threading
import time

HOST = "127.0.0.1"


def show(title: str) -> None:
    print()
    print(title)
    print("-" * len(title))


def row(label: str, value: object) -> None:
    print(f"  {label:<46} {value}")


class Service:
    """Сервис со счётом эффектов и необязательным журналом ключей.

    ЧТО СЧИТАЕТСЯ ЭФФЕКТОМ. Строка `applied += 1` — это «деньги списаны»,
    «письмо отправлено», «заказ создан»: то, что нельзя сделать дважды.
    Предмет замера — именно её счётчик, а не число полученных запросов.
    """

    def __init__(
        self,
        use_keys: bool,
        drop_reply: bool = False,
        slow_reply: float = 0.0,
        claim_key: bool = True,
    ) -> None:
        self.use_keys = use_keys
        # claim_key=False: ключ записывается ПОСЛЕ работы, а не до неё. Это и
        # есть та ошибка, которую блок 4 показывает рядом с правильным
        # порядком, а не описывает словами.
        self.claim_key = claim_key
        self.drop_reply = drop_reply
        self.slow_reply = slow_reply
        self.applied = 0
        self.seen: dict[str, str] = {}
        self.lock = threading.Lock()
        self.sock = socket.socket()
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind((HOST, 0))
        self.sock.listen(64)
        self.port = self.sock.getsockname()[1]
        threading.Thread(target=self._loop, daemon=True).start()

    def _loop(self) -> None:
        while True:
            try:
                conn, _ = self.sock.accept()
            except OSError:
                return
            threading.Thread(target=self._serve, args=(conn,), daemon=True).start()

    def _serve(self, conn: socket.socket) -> None:
        with conn:
            key = conn.recv(64).decode().strip()
            if self.use_keys:
                with self.lock:
                    done = self.seen.get(key)
                if done is not None:
                    conn.sendall(f"{done} (replayed)\n".encode())
                    return
            if self.use_keys and self.claim_key:
                with self.lock:
                    if key in self.seen:
                        conn.sendall(f"{self.seen[key]} (replayed)\n".encode())
                        return
                    self.applied += 1
                    answer = f"ok #{self.applied}"
                    self.seen[key] = answer
            else:
                with self.lock:
                    self.applied += 1
                    answer = f"ok #{self.applied}"
            if self.slow_reply:
                time.sleep(self.slow_reply)
            if self.use_keys and not self.claim_key:
                # Ключ записан после работы: между проверкой и записью есть
                # окно, в которое успевает второй запрос.
                with self.lock:
                    self.seen[key] = answer
            if self.drop_reply:
                # Ответ теряется по дороге: эффект уже случился, клиент об
                # этом не узнает и повторит.
                return
            conn.sendall((answer + "\n").encode())

    def close(self) -> None:
        self.sock.close()


def call(port: int, key: str, timeout: float = 1.0) -> str:
    client = socket.socket()
    client.settimeout(timeout)
    try:
        client.connect((HOST, port))
        client.sendall((key + "\n").encode())
        answer = client.recv(64).decode().strip()
        return answer or "empty answer"
    except socket.timeout:
        return "client timeout"
    finally:
        client.close()


# ------------------------------------------------------------------ 1 и 2
def block1() -> None:
    show("1. A RETRY WITHOUT A KEY APPLIES THE EFFECT AGAIN")

    service = Service(use_keys=False)
    answers = [call(service.port, "no-key") for _ in range(4)]
    row("requests sent", len(answers))
    row("times the effect was applied", service.applied)
    row("what the client got each time", " | ".join(answers))
    service.close()
    print()
    print("  Four identical requests, four effects. From the client's side the")
    print("  answers even look different, which is the honest picture: these")
    print("  were four separate operations, not one operation retried.")


def block2() -> None:
    show("2. THE SAME RETRIES WITH AN IDEMPOTENCY KEY")

    service = Service(use_keys=True)
    answers = [call(service.port, "order-42") for _ in range(4)]
    row("requests sent", len(answers))
    row("times the effect was applied", service.applied)
    row("what the client got each time", " | ".join(answers))
    service.close()
    print()
    print("  Four requests again, and one effect. The server did not become")
    print("  smarter: it kept the answer under the key and returned the stored")
    print("  one instead of doing the work twice.")


# ------------------------------------------------------------------ 3
def block3() -> None:
    show("3. THE WINDOW: THE EFFECT HAPPENED, THE ANSWER DID NOT ARRIVE")

    service = Service(use_keys=True, drop_reply=True)
    first = call(service.port, "order-77", timeout=0.4)
    row("what the client saw on the first try", first)
    row("times the effect was applied by then", service.applied)

    service.drop_reply = False
    second = call(service.port, "order-77")
    row("what the client saw on the retry", second)
    row("times the effect was applied in total", service.applied)
    service.close()
    print()
    print("  The client saw a failure and was right to retry: it had no way to")
    print("  know. The effect had already happened. This window cannot be")
    print("  closed - only made harmless, which is what the key does.")


# ------------------------------------------------------------------ 4
def block4() -> None:
    show("4. TWO RETRIES AT THE SAME MOMENT, ONE KEY")

    for claim_key, label in (
        (True, "key claimed before the work"),
        (False, "key written after the work"),
    ):
        service = Service(use_keys=True, slow_reply=0.05, claim_key=claim_key)
        answers: list[str] = ["", ""]

        def fire(i: int, port: int = service.port) -> None:
            answers[i] = call(port, "order-99")

        threads = [threading.Thread(target=fire, args=(i,)) for i in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()

        word = "effect" if service.applied == 1 else "effects"
        row(label, f"{service.applied} {word} from 2 requests")
        row("  answers", " | ".join(answers))
        service.close()
    print()
    print("  Same key, same two simultaneous requests, same store. The only")
    print("  difference is when the key is written: claiming it before the work")
    print("  gives one effect, writing it afterwards gives two. A store that")
    print("  merely remembers the key is not enough - it has to claim it.")


# ------------------------------------------------------------------ 5
def block5() -> None:
    show("5. A NEW KEY PER ATTEMPT PROTECTS NOTHING")

    service = Service(use_keys=True)
    before = service.applied
    for _ in range(3):
        call(service.port, "same-key")
    with_one_key = service.applied - before

    before = service.applied
    for i in range(3):
        call(service.port, f"fresh-{i}")
    with_fresh_keys = service.applied - before

    row("three attempts, one key: effects", with_one_key)
    row("three attempts, a new key each: effects", with_fresh_keys)
    row("total effects applied", service.applied)
    service.close()
    print()
    print("  The key is what says 'this is the same operation'. Three attempts")
    print("  under one key cost one effect; three attempts under three keys cost")
    print("  three - the server cannot tell them apart, and nothing about it")
    print("  changed between the two halves of this block.")


def main() -> None:
    print(f"Python {sys.version.split()[0]} · Linux {os.uname().release} · loopback")
    print("an effect here is one increment of a counter: the thing that must not happen twice")
    block1()
    block2()
    block3()
    block4()
    block5()


if __name__ == "__main__":
    main()