"""
In-Memory Concurrent LFU Cache with Async TTL Eviction and Atomic Transactions.

Pure Python 3.11+ implementation — no third-party dependencies.
"""

from __future__ import annotations

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


# ---------------------------------------------------------------------------
# Doubly-Linked List (O(1) move/remove/add)
# ---------------------------------------------------------------------------

@dataclass
class _DLLNode:
    """Element inside a frequency bucket's doubly-linked list."""
    key: Any
    value: Any
    ttl_seconds: float
    expiry_time: float
    freq: int = 0
    prev: Optional[_DLLNode] = field(default=None, repr=False)
    next: Optional[_DLLNode] = field(default=None, repr=False)


class _DoublyLinkedList:
    """
    Doubly-linked list with sentinel head/tail for O(1) prepend, pop, and remove.
    The *tail* side holds the least-recently-used node within a frequency tier,
    which is also the LFU candidate for eviction.
    """

    def __init__(self) -> None:
        self._head = _DLLNode(None, None, 0.0, 0.0)  # sentinel
        self._tail = _DLLNode(None, None, 0.0, 0.0)  # sentinel
        self._head.next = self._tail
        self._tail.prev = self._head
        self.size = 0

    # -- internal helpers ---------------------------------------------------

    def _insert_after(self, node: _DLLNode, prev: _DLLNode) -> None:
        nxt = prev.next
        prev.next = node
        node.prev = prev
        node.next = nxt
        nxt.prev = node

    def _unlink(self, node: _DLLNode) -> None:
        prev, nxt = node.prev, node.next
        prev.next = nxt
        nxt.prev = prev
        node.prev = node.next = None

    # -- public API ---------------------------------------------------------

    def push_front(self, node: _DLLNode) -> None:
        """Insert *node* right after the head sentinel (most-recent)."""
        self._insert_after(node, self._head)
        self.size += 1

    def pop_tail(self) -> Optional[_DLLNode]:
        """Remove and return the node just before the tail sentinel (LRU)."""
        if self.size == 0:
            return None
        node = self._tail.prev
        self._unlink(node)
        return node

    def remove(self, node: _DLLNode) -> None:
        """Remove an arbitrary node from the list."""
        self._unlink(node)

    def is_empty(self) -> bool:
        return self.size == 0


# ---------------------------------------------------------------------------
# Cache Node (wraps the DLL node + TTL metadata)
# ---------------------------------------------------------------------------

@dataclass
class _CacheNode:
    key: Any
    value: Any
    ttl_seconds: float
    expiry_time: float
    freq: int = 1
    dll_node: Optional[_DLLNode] = field(default=None, repr=False)


# ---------------------------------------------------------------------------
# Transaction Handle
# ---------------------------------------------------------------------------

class Transaction:
    """
    Represents an isolated sub-session on the cache.

    - ``tx.put(key, value, ttl)`` buffers a write locally.
    - ``tx.get(key)`` reads from local buffer first, then falls back to the
      global cache (without mutating global frequency state).
    - ``tx.delete(key)`` buffers a deletion locally.
    - ``await tx.commit()`` atomically applies all buffered writes to the global cache.
    - ``tx.rollback()`` discards everything.
    """

    def __init__(self, cache: "LFUCache") -> None:
        self._cache = cache
        self._pending_writes: dict[Any, tuple[Any, float]] = {}   # key -> (value, expiry)
        self._pending_deletes: set[Any] = set()
        self._committed = False

    # -- read ----------------------------------------------------------------

    def get(self, key: Any) -> Optional[Any]:
        """Read with read-your-own-writes semantics."""
        if self._committed:
            raise RuntimeError("Transaction already committed")

        # 1. Check local pending writes first
        if key in self._pending_writes:
            return self._pending_writes[key][0]
        if key in self._pending_deletes:
            return None

        # 2. Fall back to global cache (read-only, no frequency bump)
        return self._cache._get_raw(key)

    # -- write ---------------------------------------------------------------

    def put(self, key: Any, value: Any, ttl_seconds: float = 60.0) -> None:
        """Buffer a write locally; not visible to others until commit."""
        if self._committed:
            raise RuntimeError("Transaction already committed")
        expiry = time.monotonic() + ttl_seconds
        self._pending_writes[key] = (value, expiry)
        # If previously deleted in this txn, re-add overrides the delete.
        self._pending_deletes.discard(key)

    def delete(self, key: Any) -> None:
        """Buffer a deletion locally."""
        if self._committed:
            raise RuntimeError("Transaction already committed")
        self._pending_deletes.add(key)
        self._pending_writes.pop(key, None)

    # -- commit / rollback ---------------------------------------------------

    async def commit(self) -> None:
        """Atomically apply all buffered changes to the global cache."""
        if self._committed:
            raise RuntimeError("Transaction already committed")
        async with self._cache._lock:
            for key, (value, expiry) in self._pending_writes.items():
                await self._cache._put_internal(key, value, expiry)
            for key in self._pending_deletes:
                await self._cache._delete_internal(key)
        self._committed = True

    def rollback(self) -> None:
        """Discard all pending changes."""
        self._pending_writes.clear()
        self._pending_deletes.clear()
        self._committed = True  # mark so further ops raise


# ---------------------------------------------------------------------------
# LFU Cache
# ---------------------------------------------------------------------------

class LFUCache:
    """
    In-memory concurrent LFU cache with O(1) get/put, async TTL eviction,
    and atomic transaction support.
    """

    def __init__(self, capacity: int = 1024) -> None:
        if capacity < 1:
            raise ValueError("capacity must be >= 1")

        self._capacity = capacity
        self._lock = asyncio.Lock()
        self._evictor_task: Optional[asyncio.Task[None]] = None

        # Core O(1) structures
        self._cache_map: dict[Any, _CacheNode] = {}          # key -> CacheNode
        self._freq_map: dict[int, _DoublyLinkedList] = {}     # freq -> DLL
        self._min_freq: int = 1

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def get(self, key: Any) -> Optional[Any]:
        """O(1) lookup with lazy TTL eviction."""
        async with self._lock:
            node = self._cache_map.get(key)
            if node is None:
                return None
            # Lazy TTL check
            if time.monotonic() > node.expiry_time:
                await self._evict_node(key, node)
                return None
            # Bump frequency — O(1)
            await self._bump_freq(node)
            return node.value

    async def put(self, key: Any, value: Any, ttl_seconds: float = 60.0) -> None:
        """O(1) insert/update with lazy TTL eviction of LRU-LFU victim if needed."""
        expiry = time.monotonic() + ttl_seconds
        async with self._lock:
            await self._put_internal(key, value, expiry)

    async def delete(self, key: Any) -> bool:
        """O(1) deletion."""
        async with self._lock:
            return await self._delete_internal(key)

    def begin_transaction(self) -> Transaction:
        """Start a new isolated transaction."""
        return Transaction(self)

    # ------------------------------------------------------------------
    # TTL Eviction Loop
    # ------------------------------------------------------------------

    def start_evictor(self, interval_seconds: float = 1.0) -> None:
        """Start the background async TTL sweep task."""
        if self._evictor_task is not None and not self._evictor_task.done():
            return
        self._evictor_task = asyncio.create_task(self._eviction_loop(interval_seconds))

    def stop_evictor(self) -> None:
        """Stop the background eviction task."""
        if self._evictor_task is not None:
            self._evictor_task.cancel()
            self._evictor_task = None

    async def _eviction_loop(self, interval: float) -> None:
        """Periodically purge expired entries in small batches."""
        try:
            while True:
                await asyncio.sleep(interval)
                async with self._lock:
                    now = time.monotonic()
                    # Collect expired keys in a snapshot to avoid dict-changed-size
                    expired = [
                        k for k, n in self._cache_map.items()
                        if now > n.expiry_time
                    ]
                    for k in expired:
                        node = self._cache_map.get(k)
                        if node is not None and now > node.expiry_time:
                            await self._evict_node(k, node)
        except asyncio.CancelledError:
            return

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    async def _get_raw(self, key: Any) -> Optional[Any]:
        """Read-only global lookup — does NOT bump frequency. For transactions."""
        node = self._cache_map.get(key)
        if node is None:
            return None
        if time.monotonic() > node.expiry_time:
            await self._evict_node(key, node)
            return None
        return node.value

    async def _put_internal(self, key: Any, value: Any, expiry: float) -> None:
        """Core put logic (must be called under lock)."""
        # If key already exists, update in place
        if key in self._cache_map:
            node = self._cache_map[key]
            # Remove from old freq list, update value/ttl/freq
            self._freq_map[node.freq].remove(node.dll_node)
            if self._freq_map[node.freq].is_empty():
                del self._freq_map[node.freq]
                if node.freq == self._min_freq:
                    self._min_freq += 1
            node.value = value
            node.ttl_seconds = expiry - time.monotonic()
            node.expiry_time = expiry
            node.freq = 1
            self._ensure_freq_list(1).push_front(node.dll_node)
            return

        # Evict if at capacity
        if len(self._cache_map) >= self._capacity:
            await self._evict_lfu()

        # Insert new node
        dll_node = _DLLNode(key, value, expiry - time.monotonic(), expiry)
        cache_node = _CacheNode(key, value, expiry - time.monotonic(), expiry, freq=1, dll_node=dll_node)
        self._cache_map[key] = cache_node
        self._freq_map[1].push_front(dll_node)
        self._min_freq = 1

    async def _delete_internal(self, key: Any) -> bool:
        """Core delete logic (must be called under lock)."""
        node = self._cache_map.get(key)
        if node is None:
            return False
        await self._evict_node(key, node)
        return True

    async def _evict_node(self, key: Any, node: _CacheNode) -> None:
        """Remove a single node from all structures."""
        dll = self._freq_map.get(node.freq)
        if dll is not None:
            dll.remove(node.dll_node)
            if dll.is_empty():
                del self._freq_map[node.freq]
                if node.freq == self._min_freq:
                    # Find new min freq
                    new_min = min(self._freq_map) if self._freq_map else 1
                    self._min_freq = new_min
        del self._cache_map[key]

    async def _evict_lfu(self) -> None:
        """Evict the least-frequently-used (and least-recently-used within that freq) node."""
        if not self._freq_map:
            return
        dll = self._freq_map.get(self._min_freq)
        if dll is None or dll.is_empty():
            # Shouldn't happen, but safeguard
            self._min_freq += 1
            await self._evict_lfu()
            return
        victim = dll.pop_tail()
        if victim is not None:
            await self._evict_node(victim.key, self._cache_map[victim.key])

    async def _bump_freq(self, node: _CacheNode) -> None:
        """Move a node from freq f to freq f+1. O(1)."""
        old_freq = node.freq
        old_dll = self._freq_map[old_freq]
        old_dll.remove(node.dll_node)
        if old_dll.is_empty():
            del self._freq_map[old_freq]
            if old_freq == self._min_freq:
                self._min_freq += 1

        node.freq += 1
        new_dll = self._ensure_freq_list(node.freq)
        new_dll.push_front(node.dll_node)

    def _ensure_freq_list(self, freq: int) -> _DoublyLinkedList:
        """Return (or create) the DLL for *freq*."""
        if freq not in self._freq_map:
            self._freq_map[freq] = _DoublyLinkedList()
        return self._freq_map[freq]

    # ------------------------------------------------------------------
    # Diagnostics
    # ------------------------------------------------------------------

    @property
    def size(self) -> int:
        return len(self._cache_map)

    @property
    def capacity(self) -> int:
        return self._capacity


# ---------------------------------------------------------------------------
# Executable Test Suite
# ---------------------------------------------------------------------------

async def main() -> None:
    passed = 0
    failed = 0

    def _check(name: str, condition: bool) -> None:
        nonlocal passed, failed
        if condition:
            passed += 1
            print(f"  ✓ {name}")
        else:
            failed += 1
            print(f"  ✗ {name}")

    # ======================================================================
    # (a) O(1) LFU eviction order
    # ======================================================================
    print("\n=== (a) LFU Eviction Order ===")

    cache = LFUCache(capacity=3)

    # Insert 3 items
    await cache.put("a", 1, ttl_seconds=60.0)
    await cache.put("b", 2, ttl_seconds=60.0)
    await cache.put("c", 3, ttl_seconds=60.0)

    # Access "a" and "b" once each → freq=2; "c" stays at freq=1
    await cache.get("a")
    await cache.get("b")

    # Insert "d" — should evict "c" (lowest freq)
    await cache.put("d", 4, ttl_seconds=60.0)

    _check("a still present after eviction", await cache.get("a") == 1)
    _check("b still present after eviction", await cache.get("b") == 2)
    _check("c evicted (lowest freq)", await cache.get("c") is None)
    _check("d present", await cache.get("d") == 4)

    # Now access "a" again → freq=3; "b" and "d" at freq=2
    await cache.get("a")
    # Insert "e" — should evict either "b" or "d" (both freq=2, LRU wins)
    await cache.put("e", 5, ttl_seconds=60.0)

    _check("a still present", await cache.get("a") == 1)
    _check("e present", await cache.get("e") == 5)

    # ======================================================================
    # (b) Lazy TTL vs Background Async Sweep
    # ======================================================================
    print("\n=== (b) TTL Eviction (Lazy + Background) ===")

    cache2 = LFUCache(capacity=10)

    await cache2.put("lazy_key", "lazy_val", ttl_seconds=0.1)
    await cache2.put("bg_key", "bg_val", ttl_seconds=0.1)

    # Lazy eviction: access lazy_key after expiry
    await asyncio.sleep(0.15)
    _check("Lazy eviction: get returns None after TTL", await cache2.get("lazy_key") is None)
    _check("Lazy eviction: bg_key still there (not accessed)", await cache2.get("bg_key") == "bg_val")

    # Start background evictor
    cache2.start_evictor(interval_seconds=0.2)
    await asyncio.sleep(0.3)  # let background sweep run

    _check("Background eviction: bg_key purged by sweep", await cache2.get("bg_key") is None)
    cache2.stop_evictor()

    # ======================================================================
    # (c) Transaction commit visibility vs rollback
    # ======================================================================
    print("\n=== (c) Atomic Transactions ===")

    cache3 = LFUCache(capacity=10)
    await cache3.put("x", 10, ttl_seconds=60.0)
    await cache3.put("y", 20, ttl_seconds=60.0)

    # --- Commit test ---
    tx1 = cache3.begin_transaction()
    tx1.put("x", 99, ttl_seconds=60.0)   # local write
    tx1.put("z", 30, ttl_seconds=60.0)   # new key

    _check("TX: read-your-own-write (x)", tx1.get("x") == 99)
    _check("TX: read-your-own-write (z)", tx1.get("z") == 30)
    _check("TX: global still sees old x", await cache3.get("x") == 10)

    await tx1.commit()
    _check("TX: after commit, global sees x=99", await cache3.get("x") == 99)
    _check("TX: after commit, global sees z=30", await cache3.get("z") == 30)

    # --- Rollback test ---
    tx2 = cache3.begin_transaction()
    tx2.put("x", -1, ttl_seconds=60.0)
    tx2.delete("y")
    _check("TX rollback: local sees x=-1", tx2.get("x") == -1)
    _check("TX rollback: local sees y deleted", tx2.get("y") is None)
    _check("TX rollback: global still sees x=99", await cache3.get("x") == 99)
    _check("TX rollback: global still sees y=20", await cache3.get("y") == 20)

    tx2.rollback()
    _check("TX rollback: global unchanged after rollback", await cache3.get("x") == 99)
    _check("TX rollback: y still present after rollback", await cache3.get("y") == 20)

    # --- Double commit / rollback raises ---
    tx3 = cache3.begin_transaction()
    await tx3.commit()
    try:
        tx3.put("x", 1)
        _check("TX: double commit raises", False)
    except RuntimeError:
        _check("TX: double commit raises", True)

    tx4 = cache3.begin_transaction()
    tx4.rollback()
    try:
        tx4.put("x", 1)
        _check("TX: op after rollback raises", False)
    except RuntimeError:
        _check("TX: op after rollback raises", True)

    # ======================================================================
    # (d) Stress test: 50 concurrent async tasks
    # ======================================================================
    print("\n=== (d) Stress Test — 50 Concurrent Tasks ===")

    cache4 = LFUCache(capacity=200)
    errors: list[str] = []

    async def worker(task_id: int, base_key: int) -> None:
        try:
            for i in range(50):
                key = f"t{task_id}_k{i}"
                val = task_id * 1000 + i
                await cache4.put(key, val, ttl_seconds=5.0)
                result = await cache4.get(key)
                if result != val:
                    errors.append(f"task={task_id} key={key} expected={val} got={result}")
                # Occasional transaction
                if i % 10 == 0:
                    tx = cache4.begin_transaction()
                    tx.put(f"tx_{task_id}_{i}", val * 2, ttl_seconds=5.0)
                    r = tx.get(f"tx_{task_id}_{i}")
                    if r != val * 2:
                        errors.append(f"task={task_id} tx key mismatch")
                    await tx.commit()
        except Exception as e:
            errors.append(f"task={task_id} exception: {e}")

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

    _check("Stress: no errors", len(errors) == 0)
    _check("Stress: cache size within capacity", cache4.size <= 200)
    if errors:
        for e in errors[:5]:
            print(f"    error: {e}")

    # ======================================================================
    # Summary
    # ======================================================================
    total = passed + failed
    print(f"\n{'='*50}")
    print(f"Results: {passed}/{total} passed, {failed} failed")
    if failed == 0:
        print("All tests passed! ✓")
    else:
        print(f"{failed} test(s) FAILED ✗")


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