Mock LLD of in-memory cache

in_memory_cache.py

from abc import ABCMeta, abstractmethod
from collections import defaultdict
from datetime import datetime, timedelta
import hashlib
import threading

# Design philosophy
# Flow:
# Client → CacheClient → ShardRouter → CacheNode → Storage
#
# Pluggable:
# - Eviction Strategy
# - Write Strategy
# - Hashing Strategy (extensible)

# Entities

class CacheEntry(object):
    def __init__(self, key, value, ttl_seconds):
        self._key = key
        self._value = value
        self._expiry = datetime.utcnow() + timedelta(seconds=ttl_seconds)

    def get_key(self):
        return self._key

    def get_value(self):
        return self._value

    def is_expired(self):
        return datetime.utcnow() > self._expiry
      
# Eviction Strategy (Interface via ABC)


class EvictionStrategy(object, metaclass=ABCMeta):
    @abstractmethod
    def record_access(self, key):
        pass

    @abstractmethod
    def evict(self, cache_store):
        pass
      
# LRU implementation using O(1) time
from collections import OrderedDict


class LRUEvictionStrategy(EvictionStrategy):
    def __init__(self):
        self._order = OrderedDict()
        self._lock = threading.RLock()

    def record_access(self, key):
      with self._lock:
        if key in self._order:
            self._order.move_to_end(key)
        else:
            self._order[key] = True

    def evict(self, cache_store):
      with self._lock:
        if not self._order:
            return

        lru_key, _ = self._order.popitem(last=False)

        if lru_key in cache_store:
            del cache_store[lru_key]
            
# Strategy pattern to handle writes
class WriteStrategy(object, metaclass=ABCMeta):
    @abstractmethod
    def write(self, key, value, db, cache_node=None):
        pass
      
class WriteAroundStrategy(WriteStrategy):
    def write(self, key, value, db, cache_node=None):
        db[key] = value
        
class WriteThroughStrategy(WriteStrategy):
    def write(self, key, value, db, cache_node=None):
        db[key] = value

        if cache_node:
            cache_node.set(key, value)
            
# Core component - Cache node
class CacheNode(object):
    def __init__(self, node_id, capacity, eviction_strategy):
        self._node_id = node_id
        self._capacity = capacity
        self._store = dict()
        self._eviction_strategy = eviction_strategy
        self._lock = threading.RLock()

    def get(self, key):
      with self._lock:
        if key not in self._store:
            return None
          
        entry = self._store.get(key)

        if entry.is_expired():
            self._delete_internal(key)
            return None

        self._eviction_strategy.record_access(key)
        return entry.get_value()
      
    def set(self, key, value, ttl=3600):
      with self._lock:
        if len(self._store) >= self._capacity:
            self._evict()

        entry = CacheEntry(key, value, ttl)
        self._store[key] = entry

        self._eviction_strategy.record_access(key)
        
    def invalidate(self, key):
      with self._lock:
        self._delete_internal(key)
        
    def _evict(self):
        self._eviction_strategy.evict(self._store)

    def _delete_internal(self, key):
        if key in self._store:
            del self._store[key]
            
# Hashing strategy interface
class HashingStrategy(object, metaclass=ABCMeta):

    @abstractmethod
    def get_node(self, key, nodes):
        pass

class ConsistentHashingStrategy(HashingStrategy):
    def get_node(self, key, nodes):
        if not nodes:
            return None

        hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
        index = hash_val % len(nodes)

        return nodes[index]
      
class ShardRouter(object):
    def __init__(self, nodes, hashing_strategy):
        self._nodes = nodes
        self._hashing_strategy = hashing_strategy

    def route(self, key):
        return self._hashing_strategy.get_node(key, self._nodes)
      
class CacheClient(object):
    def __init__(self, shard_router, db, write_strategy):
        self._router = shard_router
        self._db = db
        self._write_strategy = write_strategy
        
    def get(self, key):
        node = self._router.route(key)

        if not node:
            return None

        value = node.get(key)

        if value is not None:
            return value

        # Cache miss → DB fallback
        value = self._db.get(key)

        if value is not None:
            node.set(key, value)

        return value
      
    def set(self, key, value, ttl=3600):
        node = self._router.route(key)

        if not node:
            return

        node.set(key, value, ttl)
        self._write_strategy.write(key, value, self._db, node)
        
    def invalidate(self, key):
        node = self._router.route(key)

        if node:
            node.invalidate(key)
            
            
# Example usecase
if __name__ == "__main__":

    # Simulated DB
    db = dict()

    # Nodes
    node1 = CacheNode("node1", capacity=3, eviction_strategy=LRUEvictionStrategy())
    node2 = CacheNode("node2", capacity=3, eviction_strategy=LRUEvictionStrategy())

    nodes = [node1, node2]

    # Router
    hashing_strategy = ConsistentHashingStrategy()
    router = ShardRouter(nodes, hashing_strategy)

    # Cache Client
    cache_client = CacheClient(
        shard_router=router,
        db=db,
        write_strategy=WriteAroundStrategy()
    )

    # Operations
    cache_client.set("user:1", "Kuntal")
    cache_client.set("user:2", "Alice")

    print(cache_client.get("user:1"))  # HIT

    cache_client.invalidate("user:1")

    print(cache_client.get("user:1"))  # MISS → DB
    
    
"""
Important pointers

“Used consistent hashing for horizontal scaling”
“Eviction is pluggable via Strategy Pattern”
“Write strategy is configurable → write-through vs write-around”
“TTL ensures bounded staleness (as required: ~48h acceptable)”
“CacheClient acts like SDK → aligns with client-side routing in doc”
"""