import asyncio
import time
from dataclasses import dataclass, field
from typing import Any, Dict, Optional


@dataclass
class Node:
    key: str
    value: Any
    ttl_expiry: float
    freq: int = 1

    def __repr__(self):
        return f"Node({self.key}, freq={self.freq})"


class DoublyLinkedListList:
    def __init__(self):
        self.head = Node("", "", 0)  # Dummy head
        self.tail = Node("", "", 0)  # Dummy tail
        self.head.next = self.tail
        self.tail.prev = self.head

    def add_to_front(self, node: Node):
        node.next = self.head.next
        node.prev = self.head
        self.head.next = node
        node.next.prev = node

    def remove(self, node: Node):
        node.prev.next = node.next
        node.next.prev = node.prev

    def move_to_front(self, node: Node):
        self.remove(node)
        self.add_to_front(node)

    def pop_tail(self) -> Optional[Node]:
        if self.tail == self.tail.prev:  # Only dummy tail left
            return None
        node = self.tail.prev
        self.remove(node)
        return node


class ConcurrentLFUCache:
    def __init__(self, capacity: int):
        if capacity <= 0:
            raise ValueError("Capacity must be positive")
        self.capacity = capacity
        self.cache: Dict[str, Node] = {}  # key -> node
        self.freq_buckets: Dict[int, DoublyLinkedListList] = {}  # freq -> list
        self.min_freq = 0
        self.global_lock = asyncio.Lock()  # Protects global cache structure
        self.evictor_task: Optional[asyncio.Task] = None

    def _update_freq(self, node: Node):
        old_freq = node.freq
        bucket = self.freq_buckets[old_freq]
        bucket.remove(node)

        if old_freq == self.min_freq and bucket.tail == bucket.head:
            self.min_freq += 1

        node.freq += 1
        if node.freq not in self.freq_buckets:
            self.freq_buckets[node.freq] = DoublyLinkedListList()

        bucket = self.freq_buckets[node.freq]
        bucket.add_to_front(node)

    async def get(self, key: str) -> Optional[Any]:
        now = time.time()
        async with self.global_lock:
            node = self.cache.get(key)
            if not node:
                return None

            if now > node.ttl_expiry:  # Lazy eviction
                bucket = self.freq_buckets[node.freq]
                bucket.remove(node)
                del self.cache[key]
                return None

            self._update_freq(node)
            return node.value

    async def put(self, key: str, value: Any, ttl_seconds: float):
        now = time.time()
        expiry = now + ttl_seconds

        async with self.global_lock:
            if key in self.cache:
                node = self.cache[key]
                bucket = self.freq_buckets[node.freq]
                bucket.remove(node)
                node.value = value
                node.ttl_expiry = expiry
                self._update_freq(node)
            else:
                if len(self.cache) >= self.capacity:
                    bucket = self.freq_buckets[self.min_freq]
                    evicted = bucket.pop_tail()
                    if evicted:
                        del self.cache[evicted.key]

                new_node = Node(key, value, expiry)
                self.cache[key] = new_node
                if self.min_freq == 0:
                    self.min_freq = 1
                bucket = self.freq_buckets[self.min_freq]
                bucket.add_to_front(new_node)

    async def start_evictor(self, interval: float = 1.0):
        def evict_loop():
            while True:
                time.sleep(interval)
                now = time.time()
                expired_keys = [k for k, n in self.cache.items() if now > n.ttl_expiry]
                for k in expired_keys:
                    asyncio.run(self.put(k, self.cache[k].value, 0))  # Re-insert with 0 TTL

        self.evictor_task = asyncio.create_task(evict_loop())

    async def stop_evictor(self):
        if self.evictor_task:
            self.evictor_task.cancel()


class Transaction:
    def __init__(self, cache):
        self.cache = cache
        self.local_cache: Dict[str, Node] = {}  # key -> node
        self.local_freq: Dict[int, DoublyLinkedListList] = {}  # freq -> list
        self.local_min_freq = 0

    def _update_local_freq(self, node: Node):
        old_freq = node.freq
        bucket = self.local_freq[old_freq]
        bucket.remove(node)

        if old_freq == self.local_min_freq and bucket.tail == bucket.head:
            self.local_min_freq += 1

        node.freq += 1
        if node.freq not in self.local_freq:
            self.local_freq[node.freq] = DoublyLinkedListList()

        bucket = self.local_freq[node.freq]
        bucket.add_to_front(node)

    async def get(self, key: str) -> Optional[Any]:
        if key in self.local_cache:
            node = self.local_cache[key]
            self._update_local_freq(node)
            return node.value

        async with self.cache.global_lock:
            node = self.cache.cache.get(key)
            if not node:
                return None

            # Copy the node so local changes don't affect global cache until commit
            new_node = Node(key, node.value, node.ttl_expiry, node.freq)
            self.local_cache[key] = new_node
            if self.local_min_freq == 0:
                self.local_min_freq = new_node.freq
            bucket = self.local_freq[new_node.freq]
            bucket.add_to_front(new_node)
            return new_node.value

    async def put(self, key: str, value: Any, ttl_seconds: float):
        now = time.time()
        expiry = now + ttl_seconds

        if key in self.local_cache:
            node = self.local_cache[key]
            bucket = self.local_freq[node.freq]
            bucket.remove(node)
            node.value = value
            node.ttl_expiry = expiry
            self._update_local_freq(node)
        else:
            if len(self.local_cache) >= self.cache.capacity:
                bucket = self.local_freq[self.local_min_freq]
                evicted = bucket.pop_tail()
                if evicted:
                    del self.local_cache[evicted.key]

            new_node = Node(key, value, expiry)
            self.local_cache[key] = new_node
            if self.local_min_freq == 0:
                self.local_min_freq = new_node.freq
            bucket = self.local_freq[new_node.freq]
            bucket.add_to_front(new_node)

    async def commit(self):
        async with self.cache.global_lock:
            for key, node in self.local_cache.items():
                if key in self.cache.cache:
                    old_node = self.cache.cache[key]
                    bucket = self.cache.freq_buckets[old_node.freq]
                    bucket.remove(old_node)
                    self.cache.cache[key] = node
                    if old_node.freq == self.cache.min_freq and bucket.tail == bucket.head:
                        self.cache.min_freq += 1
                    node.freq = node.freq  # freq is already updated in local_cache
                    bucket = self.cache.freq_buckets[node.freq]
                    if node.freq not in self.cache.freq_buckets:
                        self.cache.freq_buckets[node.freq] = DoublyLinkedListList()
                    bucket.add_to_front(node)

    async def rollback(self):
        self.local_cache = {}
        self.local_freq = {}
        self.local_min_freq = 0


async def main():
    cache = ConcurrentLFUCache(capacity=3)

    # a. O(1) LFU eviction order
    await cache.put("a", "val_a", 60)  # freq=1
    await cache.put("b", "val_b", 60)  # freq=1
    await cache.put("c", "val_c", 60)  # freq=1
    await cache.get("a")              # a:freq=2, b/c:freq=1
    await cache.get("b")              # a:freq=2, b:freq=2, c:freq=1
    await cache.put("d", "val_d", 60)  # c is evicted (lowest freq=1)
    assert await cache.get("c") is None

    # b. Lazy vs Background eviction
    await cache.put("e", "val_e", 0)   # expire immediately
    assert await cache.get("e") is None  # lazy eviction

    # c. Transaction commit vs rollback
    tx = await cache.begin_transaction()  # helper below
    await tx.put("f", "val_f", 60)
    assert await tx.get("f") == "val_f"   # Read Your Own Writes
    assert await cache.get("f") is None   # Global not yet committed

    await tx.rollback()
    assert await cache.get("f") is None   # Rollback restored

    await tx.commit()
    assert await cache.get("f") == "val_f"  # Committed

    # d. Stress test
    async def worker(i):
        await cache.put(f"key_{i}", i, 10)
        await cache.get(f"key_{i}")

    tasks = [worker(i) for i in range(50)]
    await asyncio.gather(*tasks)

def begin_transaction():  # Helper for the test suite
    return Transaction(ConcurrentLFUCache(capacity=3))


if __name__ == "__main__":
    asyncio.run(main())