"""Reference solution — Versioned Key-Value Store.

DO NOT READ BEFORE YOU HAVE RUN THE PROBLEM UNDER THE CLOCK.

--------------------------------------------------------------------------
The representation that survives all four gates
--------------------------------------------------------------------------

The tempting gate-1 design is `dict[key] -> dict[version] -> value`, and it
passes gate 1. It dies at gate 2, because "the value as of version v" needs
the LARGEST version <= v, and a hash map cannot answer that without scanning.

The representation that survives is per key an APPEND-ONLY LIST of
(version, value) ordered by version, searched with bisect:

    _data[key] = [(v1, value1), (v2, value2), ...]     strictly increasing

    get at v      bisect_right for the last entry <= v          O(log n)
    put           append                                        O(1)
    delete        append a TOMBSTONE                            O(1)
    history       the list itself                               O(1)
    snapshot      pin a version integer                         O(1)
    compact       keep only entries visible from a live pin     O(n)

Two decisions do the heavy lifting, and both are worth saying out loud:

1. DELETE IS A WRITE. A tombstone, not a removal. Removing the key would
   destroy the ability to read at an earlier version, which is the entire
   product. This is why delete() consumes a version even for a key that does
   not exist -- versions are a property of the log, not of the data.

2. VERSIONS ARE GLOBAL, not per key. That is what makes a snapshot a single
   integer and makes cross-key transactions possible at all. Per-key
   versions would force a vector clock to express "read everything as of
   now", and every later gate would be harder.

This is MVCC. Gate 4 adds optimistic concurrency control on top: read at a
pinned version, track the read set, and validate at commit. What you get is
SNAPSHOT ISOLATION -- so be ready for the follow-up: it still permits WRITE
SKEW, because two transactions can read overlapping data, write disjoint
keys, and both commit while jointly violating an invariant neither one
violated alone.
"""

from __future__ import annotations

from bisect import bisect_right
from typing import Any


class _Deleted:
    """Tombstone sentinel. A singleton so `is` comparisons work."""

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __repr__(self) -> str:
        return "DELETED"


DELETED = _Deleted()


class ConflictError(Exception):
    """Raised when a transaction's read set was written under it."""


class Snapshot:
    """A stable read view pinned to a version."""

    __slots__ = ("_store", "version", "_released")

    def __init__(self, store: "VersionedKV", version: int) -> None:
        self._store = store
        self.version = version
        self._released = False

    def get(self, key: str) -> Any:
        if self._released:
            raise RuntimeError("snapshot has been released")
        return self._store.get(key, version=self.version)

    def keys(self) -> list[str]:
        if self._released:
            raise RuntimeError("snapshot has been released")
        return self._store.keys(version=self.version)

    def release(self) -> None:
        if self._released:
            return
        self._released = True
        self._store._release_snapshot(self)

    def __enter__(self) -> "Snapshot":
        return self

    def __exit__(self, *exc) -> None:
        self.release()


class Transaction:
    """Optimistic, snapshot-isolated multi-key transaction."""

    __slots__ = ("_store", "_read_version", "_reads", "_writes", "_done")

    def __init__(self, store: "VersionedKV") -> None:
        self._store = store
        self._read_version = store.version
        self._reads: set[str] = set()
        self._writes: dict[str, Any] = {}
        self._done = False

    def _check_open(self) -> None:
        if self._done:
            raise RuntimeError("transaction already finished")

    def get(self, key: str) -> Any:
        self._check_open()
        self._reads.add(key)
        if key in self._writes:  # read-your-own-writes
            value = self._writes[key]
            return None if value is DELETED else value
        return self._store.get(key, version=self._read_version)

    def put(self, key: str, value: Any) -> None:
        self._check_open()
        self._writes[key] = value

    def delete(self, key: str) -> None:
        self._check_open()
        self._writes[key] = DELETED

    def rollback(self) -> None:
        self._done = True
        self._writes.clear()

    def commit(self) -> int:
        self._check_open()
        # Validate: nobody may have written anything we READ since we started.
        for key in self._reads:
            if self._store._last_write_version(key) > self._read_version:
                self._done = True
                raise ConflictError(
                    f"key {key!r} was written at version "
                    f"{self._store._last_write_version(key)} after this "
                    f"transaction read at version {self._read_version}"
                )
        self._done = True
        if not self._writes:
            return self._store.version
        # Atomic: every write in the transaction shares one version, so no
        # reader can ever observe a partial transaction.
        return self._store._apply_batch(self._writes)


class VersionedKV:
    """A multi-version key-value store with point-in-time reads."""

    __slots__ = ("_data", "_version", "_live_snapshots")

    def __init__(self) -> None:
        self._data: dict[str, list[tuple[int, Any]]] = {}
        self._version = 0
        self._live_snapshots: list[Snapshot] = []

    # --- gate 1 ------------------------------------------------------------

    @property
    def version(self) -> int:
        return self._version

    def put(self, key: str, value: Any) -> int:
        self._version += 1
        self._data.setdefault(key, []).append((self._version, value))
        return self._version

    def get(self, key: str, version: int | None = None) -> Any:
        entry = self._entry_at(key, self._version if version is None else version)
        if entry is None or entry[1] is DELETED:
            return None
        return entry[1]

    def _entry_at(self, key: str, version: int) -> tuple[int, Any] | None:
        entries = self._data.get(key)
        if not entries:
            return None
        # Entries are strictly increasing in version, so bisect on the version
        # column finds the last write at or before `version` in O(log n).
        index = bisect_right(entries, version, key=lambda item: item[0])
        return entries[index - 1] if index else None

    # --- gate 2 ------------------------------------------------------------

    def delete(self, key: str) -> int:
        # A tombstone, not a removal: an earlier read must still see the old
        # value. Consumes a version even for an absent key, because versions
        # describe the log, not the data.
        self._version += 1
        self._data.setdefault(key, []).append((self._version, DELETED))
        return self._version

    def history(self, key: str) -> list[tuple[int, Any]]:
        return list(self._data.get(key, ()))

    def keys(self, version: int | None = None) -> list[str]:
        at = self._version if version is None else version
        live = []
        for key in self._data:
            entry = self._entry_at(key, at)
            if entry is not None and entry[1] is not DELETED:
                live.append(key)
        return sorted(live)

    def __contains__(self, key: str) -> bool:
        return self.get(key) is not None

    # --- gate 3 ------------------------------------------------------------

    def snapshot(self) -> Snapshot:
        snap = Snapshot(self, self._version)
        self._live_snapshots.append(snap)
        return snap

    def _release_snapshot(self, snap: Snapshot) -> None:
        try:
            self._live_snapshots.remove(snap)
        except ValueError:
            pass

    def compact(self) -> int:
        """Drop versions no live reader can ever observe. Returns the count."""
        # Every version that some reader can still land on is a "pin": the
        # current version, plus every live snapshot's version. For each key we
        # must retain exactly the entry visible from each pin; everything else
        # is unreachable and can go.
        pins = {self._version}
        pins.update(snap.version for snap in self._live_snapshots)

        dropped = 0
        for key, entries in list(self._data.items()):
            if len(entries) <= 1:
                continue
            keep_indices = set()
            for pin in pins:
                index = bisect_right(entries, pin, key=lambda item: item[0])
                if index:
                    keep_indices.add(index - 1)
            if len(keep_indices) == len(entries):
                continue
            retained = [entries[i] for i in sorted(keep_indices)]
            dropped += len(entries) - len(retained)
            if retained:
                self._data[key] = retained
            else:
                # No pin can see this key at all — it predates every reader.
                del self._data[key]
                dropped += 0
        return dropped

    # --- gate 4 ------------------------------------------------------------

    def begin(self) -> Transaction:
        return Transaction(self)

    def _last_write_version(self, key: str) -> int:
        entries = self._data.get(key)
        return entries[-1][0] if entries else 0

    def _apply_batch(self, writes: dict[str, Any]) -> int:
        self._version += 1
        for key, value in writes.items():
            self._data.setdefault(key, []).append((self._version, value))
        return self._version