// Load balancer implementation with health checks
class LoadBalancer {
  private servers: Server[] = [];
  private currentIndex = 0;
  private healthCheckInterval: NodeJS.Timer;

  constructor(serverAddresses: string[]) {
    this.servers = serverAddresses.map(
      addr => new Server(addr)
    );
    this.startHealthChecks();
  }

  private startHealthChecks(): void {
    this.healthCheckInterval = setInterval(() => {
      this.servers.forEach(server => {
        this.checkServerHealth(server);
      });
    }, 5000); // Check every 5 seconds
  }

  private async checkServerHealth(server: Server): Promise<void> {
    try {
      const response = await fetch(`http://${server.address}/health`, {
        timeout: 2000,
      });
      server.healthy = response.ok;
    } catch {
      server.healthy = false;
    }
  }

  // Least connections algorithm
  selectServer(): Server {
    const healthyServers = this.servers.filter(s => s.healthy);
    
    if (healthyServers.length === 0) {
      throw new Error('No healthy servers available');
    }

    return healthyServers.reduce((prev, current) =>
      prev.activeConnections < current.activeConnections ? prev : current
    );
  }

  async handleRequest(request: any): Promise<any> {
    const server = this.selectServer();
    server.activeConnections++;

    try {
      const response = await fetch(`http://${server.address}${request.path}`, {
        method: request.method,
        headers: request.headers,
        body: request.body,
      });
      return response;
    } finally {
      server.activeConnections--;
    }
  }
}

class Server {
  address: string;
  healthy: boolean = true;
  activeConnections: number = 0;

  constructor(address: string) {
    this.address = address;
  }
}

import org.springframework.stereotype.Component;
import java.net.URI;
import java.net.http.*;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class LoadBalancer {

    private final List<Server> servers = new CopyOnWriteArrayList<>();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private final HttpClient httpClient = HttpClient.newBuilder()
        .connectTimeout(Duration.ofMillis(2000))
        .build();

    public LoadBalancer(List<String> serverAddresses) {
        serverAddresses.forEach(addr -> servers.add(new Server(addr)));
        startHealthChecks();
    }

    private void startHealthChecks() {
        scheduler.scheduleAtFixedRate(() ->
            servers.forEach(this::checkServerHealth), 0, 5, TimeUnit.SECONDS);
    }

    private void checkServerHealth(Server server) {
        try {
            var request = HttpRequest.newBuilder()
                .uri(URI.create("http://" + server.address + "/health"))
                .timeout(Duration.ofMillis(2000))
                .GET().build();
            var response = httpClient.send(request, HttpResponse.BodyHandlers.discarding());
            server.healthy = response.statusCode() == 200;
        } catch (Exception e) {
            server.healthy = false;
        }
    }

    // Least connections algorithm
    public Server selectServer() {
        return servers.stream()
            .filter(s -> s.healthy)
            .min(Comparator.comparingInt(s -> s.activeConnections.get()))
            .orElseThrow(() -> new IllegalStateException("No healthy servers available"));
    }

    public HttpResponse<String> handleRequest(String path, String method) throws Exception {
        Server server = selectServer();
        server.activeConnections.incrementAndGet();
        try {
            var request = HttpRequest.newBuilder()
                .uri(URI.create("http://" + server.address + path))
                .method(method, HttpRequest.BodyPublishers.noBody())
                .build();
            return httpClient.send(request, HttpResponse.BodyHandlers.ofString());
        } finally {
            server.activeConnections.decrementAndGet();
        }
    }

    public static class Server {
        final String address;
        volatile boolean healthy = true;
        final AtomicInteger activeConnections = new AtomicInteger(0);

        Server(String address) { this.address = address; }
    }
}

import asyncio
import aiohttp
from dataclasses import dataclass, field

@dataclass
class Server:
    address: str
    healthy: bool = True
    active_connections: int = 0


class LoadBalancer:
    def __init__(self, server_addresses: list[str]):
        self.servers = [Server(addr) for addr in server_addresses]
        self._health_task: asyncio.Task | None = None

    async def start(self):
        self._health_task = asyncio.create_task(self._health_check_loop())

    async def _health_check_loop(self):
        while True:
            await asyncio.gather(
                *[self._check_server_health(s) for s in self.servers],
                return_exceptions=True,
            )
            await asyncio.sleep(5)

    async def _check_server_health(self, server: Server) -> None:
        try:
            async with aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=2)
            ) as session:
                async with session.get(f"http://{server.address}/health") as resp:
                    server.healthy = resp.ok
        except Exception:
            server.healthy = False

    def select_server(self) -> Server:
        healthy = [s for s in self.servers if s.healthy]
        if not healthy:
            raise RuntimeError("No healthy servers available")
        # Least connections
        return min(healthy, key=lambda s: s.active_connections)

    async def handle_request(self, path: str, method: str = "GET") -> aiohttp.ClientResponse:
        server = self.select_server()
        server.active_connections += 1
        try:
            async with aiohttp.ClientSession() as session:
                async with session.request(method, f"http://{server.address}{path}") as resp:
                    return resp
        finally:
            server.active_connections -= 1

using System.Collections.Concurrent;

public class LoadBalancer
{
    private readonly List<Server> _servers;
    private readonly HttpClient _httpClient;
    private readonly Timer _healthCheckTimer;

    public LoadBalancer(IEnumerable<string> serverAddresses)
    {
        _servers = serverAddresses.Select(a => new Server(a)).ToList();
        _httpClient = new HttpClient { Timeout = TimeSpan.FromMilliseconds(2000) };
        _healthCheckTimer = new Timer(_ => CheckAllServers(), null,
            TimeSpan.Zero, TimeSpan.FromSeconds(5));
    }

    private void CheckAllServers()
    {
        foreach (var server in _servers)
            _ = CheckServerHealthAsync(server);
    }

    private async Task CheckServerHealthAsync(Server server)
    {
        try
        {
            var response = await _httpClient.GetAsync($"http://{server.Address}/health");
            server.Healthy = response.IsSuccessStatusCode;
        }
        catch
        {
            server.Healthy = false;
        }
    }

    // Least connections algorithm
    public Server SelectServer()
    {
        var healthy = _servers.Where(s => s.Healthy).ToList();
        if (healthy.Count == 0)
            throw new InvalidOperationException("No healthy servers available");

        return healthy.MinBy(s => s.ActiveConnections)!;
    }

    public async Task<HttpResponseMessage> HandleRequestAsync(
        string path, HttpMethod method)
    {
        var server = SelectServer();
        Interlocked.Increment(ref server.ActiveConnectionsField);
        try
        {
            var request = new HttpRequestMessage(method,
                $"http://{server.Address}{path}");
            return await _httpClient.SendAsync(request);
        }
        finally
        {
            Interlocked.Decrement(ref server.ActiveConnectionsField);
        }
    }
}

public class Server
{
    public string Address { get; }
    public volatile bool Healthy = true;
    public int ActiveConnectionsField;
    public int ActiveConnections => ActiveConnectionsField;

    public Server(string address) => Address = address;
}

// Cache-aside pattern with Redis
import redis from 'redis';

const client = redis.createClient({ host: 'localhost', port: 6379 });

async function getUser(userId: string): Promise<User> {
  // Check Redis first (L3 cache)
  const cached = await client.get(`user:${userId}`);
  
  if (cached) {
    return JSON.parse(cached);
  }

  // Cache miss: query database
  const user = await database.query('SELECT * FROM users WHERE id = ?', [userId]);
  
  // Store in cache for 1 hour
  await client.setex(`user:${userId}`, 3600, JSON.stringify(user));
  
  return user;
}

// Cache invalidation on updates
async function updateUser(userId: string, updates: Partial<User>): Promise<User> {
  const user = await database.query('UPDATE users SET ? WHERE id = ?', [
    updates,
    userId,
  ]);

  // Invalidate cache
  await client.del(`user:${userId}`);

  return user;
}

// Batch cache warming
async function warmCache(): Promise<void> {
  const users = await database.query('SELECT * FROM users LIMIT 1000');
  
  for (const user of users) {
    await client.setex(
      `user:${user.id}`,
      3600,
      JSON.stringify(user)
    );
  }
}

import org.springframework.cache.annotation.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.Duration;
import java.util.List;

@Service
public class UserCacheService {

    private final StringRedisTemplate redis;
    private final UserRepository userRepository;
    private final ObjectMapper objectMapper;

    public UserCacheService(StringRedisTemplate redis,
                             UserRepository userRepository,
                             ObjectMapper objectMapper) {
        this.redis = redis;
        this.userRepository = userRepository;
        this.objectMapper = objectMapper;
    }

    public User getUser(String userId) throws Exception {
        // Check Redis first (L3 cache)
        String cached = redis.opsForValue().get("user:" + userId);
        if (cached != null) {
            return objectMapper.readValue(cached, User.class);
        }

        // Cache miss: query database
        User user = userRepository.findById(userId)
            .orElseThrow(() -> new RuntimeException("User not found"));

        // Store in cache for 1 hour
        redis.opsForValue().set(
            "user:" + userId,
            objectMapper.writeValueAsString(user),
            Duration.ofHours(1)
        );

        return user;
    }

    public User updateUser(String userId, UserUpdates updates) throws Exception {
        User user = userRepository.update(userId, updates);
        // Invalidate cache
        redis.delete("user:" + userId);
        return user;
    }

    public void warmCache() throws Exception {
        List<User> users = userRepository.findTop1000();
        for (User user : users) {
            redis.opsForValue().set(
                "user:" + user.getId(),
                objectMapper.writeValueAsString(user),
                Duration.ofHours(1)
            );
        }
    }
}

import redis
import json
from dataclasses import dataclass, asdict

redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)


@dataclass
class User:
    id: str
    email: str
    name: str


def get_user(user_id: str) -> User:
    # Check Redis first (L3 cache)
    cached = redis_client.get(f"user:{user_id}")
    if cached:
        return User(**json.loads(cached))

    # Cache miss: query database
    user = database_query("SELECT * FROM users WHERE id = %s", [user_id])

    # Store in cache for 1 hour
    redis_client.setex(f"user:{user_id}", 3600, json.dumps(asdict(user)))
    return user


def update_user(user_id: str, updates: dict) -> User:
    user = database_query("UPDATE users SET ... WHERE id = %s", [user_id])
    # Invalidate cache
    redis_client.delete(f"user:{user_id}")
    return user


def warm_cache() -> None:
    users = database_query("SELECT * FROM users LIMIT 1000")
    pipeline = redis_client.pipeline()
    for user in users:
        pipeline.setex(f"user:{user.id}", 3600, json.dumps(asdict(user)))
    pipeline.execute()


def database_query(sql: str, params: list = None):
    # Placeholder for actual DB call
    raise NotImplementedError

using Microsoft.Extensions.Caching.Distributed;
using System.Text.Json;

public class UserCacheService
{
    private readonly IDistributedCache _cache;
    private readonly IUserRepository _userRepository;

    public UserCacheService(IDistributedCache cache, IUserRepository userRepository)
    {
        _cache = cache;
        _userRepository = userRepository;
    }

    public async Task<User> GetUserAsync(string userId)
    {
        // Check Redis first (L3 cache)
        var cached = await _cache.GetStringAsync($"user:{userId}");
        if (cached is not null)
            return JsonSerializer.Deserialize<User>(cached)!;

        // Cache miss: query database
        var user = await _userRepository.FindByIdAsync(userId)
            ?? throw new KeyNotFoundException($"User {userId} not found");

        // Store in cache for 1 hour
        var options = new DistributedCacheEntryOptions
        {
            AbsoluteExpirationRelativeToNow = TimeSpan.FromHours(1)
        };
        await _cache.SetStringAsync(
            $"user:{userId}", JsonSerializer.Serialize(user), options);

        return user;
    }

    public async Task<User> UpdateUserAsync(string userId, UserUpdates updates)
    {
        var user = await _userRepository.UpdateAsync(userId, updates);
        // Invalidate cache
        await _cache.RemoveAsync($"user:{userId}");
        return user;
    }

    public async Task WarmCacheAsync()
    {
        var users = await _userRepository.FindTop1000Async();
        var options = new DistributedCacheEntryOptions
        {
            AbsoluteExpirationRelativeToNow = TimeSpan.FromHours(1)
        };

        var tasks = users.Select(user =>
            _cache.SetStringAsync(
                $"user:{user.Id}", JsonSerializer.Serialize(user), options));

        await Task.WhenAll(tasks);
    }
}

// Custom cache implementation with TTL
class ExpiringCache<K, V> {
  private cache: Map<K, { value: V; expiresAt: number }> = new Map();

  set(key: K, value: V, ttlSeconds: number): void {
    const expiresAt = Date.now() + ttlSeconds * 1000;
    this.cache.set(key, { value, expiresAt });
  }

  get(key: K): V | null {
    const item = this.cache.get(key);
    
    if (!item) return null;
    
    if (Date.now() > item.expiresAt) {
      this.cache.delete(key);
      return null;
    }

    return item.value;
  }

  // Cleanup expired entries periodically
  cleanup(): void {
    const now = Date.now();
    const expiredKeys: K[] = [];

    for (const [key, item] of this.cache.entries()) {
      if (now > item.expiresAt) {
        expiredKeys.push(key);
      }
    }

    expiredKeys.forEach(key => this.cache.delete(key));
  }
}

import java.util.Map;
import java.util.concurrent.*;

public class ExpiringCache<K, V> {

    private record Entry<V>(V value, long expiresAt) {}

    private final ConcurrentHashMap<K, Entry<V>> cache = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public ExpiringCache() {
        // Cleanup expired entries every 60 seconds
        scheduler.scheduleAtFixedRate(this::cleanup, 60, 60, TimeUnit.SECONDS);
    }

    public void set(K key, V value, long ttlSeconds) {
        long expiresAt = System.currentTimeMillis() + ttlSeconds * 1000L;
        cache.put(key, new Entry<>(value, expiresAt));
    }

    public V get(K key) {
        Entry<V> entry = cache.get(key);
        if (entry == null) return null;

        if (System.currentTimeMillis() > entry.expiresAt()) {
            cache.remove(key);
            return null;
        }

        return entry.value();
    }

    public void cleanup() {
        long now = System.currentTimeMillis();
        cache.entrySet().removeIf(e -> now > e.getValue().expiresAt());
    }

    public void shutdown() {
        scheduler.shutdown();
    }
}

import time
import threading
from typing import Generic, TypeVar

K = TypeVar("K")
V = TypeVar("V")


class ExpiringCache(Generic[K, V]):
    def __init__(self):
        self._cache: dict[K, tuple[V, float]] = {}
        self._lock = threading.Lock()

    def set(self, key: K, value: V, ttl_seconds: int) -> None:
        expires_at = time.time() + ttl_seconds
        with self._lock:
            self._cache[key] = (value, expires_at)

    def get(self, key: K) -> V | None:
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                return None
            value, expires_at = entry
            if time.time() > expires_at:
                del self._cache[key]
                return None
            return value

    def cleanup(self) -> None:
        now = time.time()
        with self._lock:
            expired = [k for k, (_, exp) in self._cache.items() if now > exp]
            for key in expired:
                del self._cache[key]

using System.Collections.Concurrent;

public class ExpiringCache<K, V> where K : notnull
{
    private record Entry(V Value, long ExpiresAt);

    private readonly ConcurrentDictionary<K, Entry> _cache = new();
    private readonly Timer _cleanupTimer;

    public ExpiringCache()
    {
        _cleanupTimer = new Timer(_ => Cleanup(), null,
            TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(60));
    }

    public void Set(K key, V value, int ttlSeconds)
    {
        var expiresAt = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() + ttlSeconds * 1000L;
        _cache[key] = new Entry(value, expiresAt);
    }

    public V? Get(K key)
    {
        if (!_cache.TryGetValue(key, out var entry))
            return default;

        if (DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() > entry.ExpiresAt)
        {
            _cache.TryRemove(key, out _);
            return default;
        }

        return entry.Value;
    }

    public void Cleanup()
    {
        var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
        foreach (var key in _cache.Keys)
        {
            if (_cache.TryGetValue(key, out var entry) && now > entry.ExpiresAt)
                _cache.TryRemove(key, out _);
        }
    }
}

// Shard routing with consistent hashing
class ShardRouter {
  private shards: Map<string, Database> = new Map();
  private ring: string[] = [];

  constructor(shardAddresses: string[]) {
    shardAddresses.forEach(addr => {
      this.shards.set(addr, new Database(addr));
      // Add multiple points for load distribution
      for (let i = 0; i < 160; i++) { // 160 virtual nodes
        this.ring.push(`${addr}:${i}`);
      }
    });
    this.ring.sort();
  }

  private hashKey(key: string): number {
    let hash = 0;
    for (let i = 0; i < key.length; i++) {
      hash = ((hash << 5) - hash) + key.charCodeAt(i);
      hash |= 0; // Convert to 32-bit integer
    }
    return Math.abs(hash);
  }

  getShard(key: string): Database {
    const hash = this.hashKey(key);
    const hashRing = this.ring.map((node, idx) => ({
      node,
      position: idx,
    }));

    // Find the first shard with hash >= key hash
    const shard = hashRing.find(s => s.position >= hash % this.ring.length);
    const shardAddress = (shard || hashRing[0]).node.split(':')[0];

    return this.shards.get(shardAddress)!;
  }

  async getUserData(userId: string): Promise<any> {
    const shard = this.getShard(userId);
    return shard.query('SELECT * FROM users WHERE id = ?', [userId]);
  }

  async updateUserData(userId: string, data: any): Promise<void> {
    const shard = this.getShard(userId);
    await shard.query('UPDATE users SET ? WHERE id = ?', [data, userId]);
  }
}

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.*;

public class ShardRouter {

    private final TreeMap<Long, String> ring = new TreeMap<>();
    private final Map<String, DataSource> shards = new HashMap<>();
    private static final int VIRTUAL_NODES = 160;

    public ShardRouter(List<String> shardAddresses) {
        for (String addr : shardAddresses) {
            shards.put(addr, createDataSource(addr));
            for (int i = 0; i < VIRTUAL_NODES; i++) {
                long hash = hashKey(addr + ":" + i);
                ring.put(hash, addr);
            }
        }
    }

    private long hashKey(String key) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] bytes = md.digest(key.getBytes(StandardCharsets.UTF_8));
            return ((long)(bytes[3] & 0xFF) << 24)
                 | ((long)(bytes[2] & 0xFF) << 16)
                 | ((long)(bytes[1] & 0xFF) << 8)
                 |  (long)(bytes[0] & 0xFF);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public DataSource getShard(String key) {
        long hash = hashKey(key);
        Map.Entry<Long, String> entry = ring.ceilingEntry(hash);
        if (entry == null) entry = ring.firstEntry();
        return shards.get(entry.getValue());
    }

    public Object getUserData(String userId) throws Exception {
        DataSource shard = getShard(userId);
        try (var conn = shard.getConnection();
             var stmt = conn.prepareStatement("SELECT * FROM users WHERE id = ?")) {
            stmt.setString(1, userId);
            return stmt.executeQuery();
        }
    }

    public void updateUserData(String userId, Map<String, Object> data) throws Exception {
        DataSource shard = getShard(userId);
        // Execute update on correct shard
    }

    private DataSource createDataSource(String addr) {
        // Configure and return a DataSource for the shard address
        throw new UnsupportedOperationException("Configure actual DataSource");
    }
}

import hashlib
from dataclasses import dataclass, field


@dataclass
class ShardRouter:
    _ring: dict[int, str] = field(default_factory=dict)
    _shards: dict[str, object] = field(default_factory=dict)
    _sorted_keys: list[int] = field(default_factory=list)

    VIRTUAL_NODES = 160

    def __init__(self, shard_addresses: list[str]):
        self._ring = {}
        self._shards = {}

        for addr in shard_addresses:
            self._shards[addr] = self._create_db_connection(addr)
            for i in range(self.VIRTUAL_NODES):
                hash_val = self._hash_key(f"{addr}:{i}")
                self._ring[hash_val] = addr

        self._sorted_keys = sorted(self._ring.keys())

    def _hash_key(self, key: str) -> int:
        digest = hashlib.md5(key.encode()).digest()
        return int.from_bytes(digest[:4], "little")

    def get_shard(self, key: str):
        h = self._hash_key(key)
        for ring_key in self._sorted_keys:
            if ring_key >= h:
                return self._shards[self._ring[ring_key]]
        return self._shards[self._ring[self._sorted_keys[0]]]

    def get_user_data(self, user_id: str) -> dict:
        shard = self.get_shard(user_id)
        return shard.query("SELECT * FROM users WHERE id = %s", [user_id])

    def update_user_data(self, user_id: str, data: dict) -> None:
        shard = self.get_shard(user_id)
        shard.execute("UPDATE users SET ... WHERE id = %s", [user_id])

    def _create_db_connection(self, addr: str):
        # Return actual DB connection/session factory
        raise NotImplementedError

using System.Security.Cryptography;
using System.Text;

public class ShardRouter
{
    private readonly SortedDictionary<uint, string> _ring = new();
    private readonly Dictionary<string, IDbConnection> _shards = new();
    private const int VirtualNodes = 160;

    public ShardRouter(IEnumerable<string> shardAddresses)
    {
        foreach (var addr in shardAddresses)
        {
            _shards[addr] = CreateConnection(addr);
            for (int i = 0; i < VirtualNodes; i++)
            {
                var hash = HashKey($"{addr}:{i}");
                _ring.TryAdd(hash, addr);
            }
        }
    }

    private static uint HashKey(string key)
    {
        var bytes = MD5.HashData(Encoding.UTF8.GetBytes(key));
        return BitConverter.ToUInt32(bytes, 0);
    }

    public IDbConnection GetShard(string key)
    {
        var hash = HashKey(key);
        var entry = _ring.FirstOrDefault(kv => kv.Key >= hash);
        var addr = entry.Equals(default(KeyValuePair<uint, string>))
            ? _ring.First().Value
            : entry.Value;
        return _shards[addr];
    }

    public async Task<object?> GetUserDataAsync(string userId)
    {
        var shard = GetShard(userId);
        // Execute query on correct shard
        using var cmd = shard.CreateCommand();
        cmd.CommandText = "SELECT * FROM users WHERE id = @id";
        // add parameter and execute...
        return await Task.FromResult<object?>(null);
    }

    public async Task UpdateUserDataAsync(string userId, object data)
    {
        var shard = GetShard(userId);
        // Execute update on correct shard
        await Task.CompletedTask;
    }

    private IDbConnection CreateConnection(string addr)
    {
        // Configure and return actual DB connection
        throw new NotImplementedException("Configure actual DB connection");
    }
}

// Message queue pattern with RabbitMQ
import amqp from 'amqplib';

class MessageQueue {
  private connection: any;
  private channel: any;

  async connect(): Promise<void> {
    this.connection = await amqp.connect('amqp://localhost');
    this.channel = await this.connection.createChannel();
  }

  // Producer: send message
  async publishEvent(eventType: string, payload: any): Promise<void> {
    const exchange = 'events';
    
    await this.channel.assertExchange(exchange, 'topic', { durable: true });
    
    const message = JSON.stringify({
      type: eventType,
      timestamp: new Date(),
      payload,
    });

    this.channel.publish(
      exchange,
      eventType,
      Buffer.from(message),
      { persistent: true }
    );

    console.log(`Published event: ${eventType}`);
  }

  // Consumer: subscribe to events
  async subscribeToEvent(
    eventType: string,
    handler: (message: any) => Promise<void>
  ): Promise<void> {
    const exchange = 'events';
    const queue = `${eventType}_queue`;

    await this.channel.assertExchange(exchange, 'topic', { durable: true });
    await this.channel.assertQueue(queue, { durable: true });
    await this.channel.bindQueue(queue, exchange, eventType);

    // Prefetch one message at a time
    await this.channel.prefetch(1);

    this.channel.consume(queue, async (msg: any) => {
      if (msg) {
        try {
          const content = JSON.parse(msg.content.toString());
          await handler(content);
          this.channel.ack(msg);
        } catch (error) {
          console.error('Error processing message:', error);
          // Nack and requeue
          this.channel.nack(msg, false, true);
        }
      }
    });
  }

  async close(): Promise<void> {
    await this.channel.close();
    await this.connection.close();
  }
}

// Usage
const mq = new MessageQueue();
await mq.connect();

// Producer
await mq.publishEvent('user.created', { userId: 123, email: 'user@example.com' });

// Consumer
await mq.subscribeToEvent('user.created', async (event) => {
  console.log('New user created:', event.payload);
  // Send welcome email, initialize user profile, etc.
});

import com.rabbitmq.client.*;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.core.*;
import org.springframework.stereotype.*;
import java.util.Map;

@Service
public class MessageQueueService {

    private final RabbitTemplate rabbitTemplate;
    private static final String EXCHANGE = "events";

    public MessageQueueService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    // Producer: publish event
    public void publishEvent(String eventType, Object payload) {
        var message = Map.of(
            "type", eventType,
            "timestamp", java.time.Instant.now().toString(),
            "payload", payload
        );
        rabbitTemplate.convertAndSend(EXCHANGE, eventType, message);
        System.out.println("Published event: " + eventType);
    }
}

// Consumer: subscribe with Spring AMQP annotation
@Component
class UserEventConsumer {

    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "user.created_queue", durable = "true"),
        exchange = @Exchange(value = "events", type = ExchangeTypes.TOPIC),
        key = "user.created"
    ))
    public void handleUserCreated(Map<String, Object> event) {
        System.out.println("New user created: " + event.get("payload"));
        // Send welcome email, initialize user profile, etc.
    }
}

import aio_pika
import json
from collections.abc import Callable, Awaitable
from datetime import datetime, timezone


class MessageQueue:
    def __init__(self):
        self.connection = None
        self.channel = None

    async def connect(self) -> None:
        self.connection = await aio_pika.connect_robust("amqp://localhost")
        self.channel = await self.connection.channel()

    async def publish_event(self, event_type: str, payload: dict) -> None:
        exchange = await self.channel.declare_exchange(
            "events", aio_pika.ExchangeType.TOPIC, durable=True
        )
        message_body = json.dumps({
            "type": event_type,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "payload": payload,
        })
        await exchange.publish(
            aio_pika.Message(
                body=message_body.encode(),
                delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
            ),
            routing_key=event_type,
        )
        print(f"Published event: {event_type}")

    async def subscribe_to_event(
        self,
        event_type: str,
        handler: Callable[[dict], Awaitable[None]],
    ) -> None:
        exchange = await self.channel.declare_exchange(
            "events", aio_pika.ExchangeType.TOPIC, durable=True
        )
        queue = await self.channel.declare_queue(
            f"{event_type}_queue", durable=True
        )
        await queue.bind(exchange, routing_key=event_type)
        await self.channel.set_qos(prefetch_count=1)

        async def on_message(message: aio_pika.IncomingMessage):
            async with message.process(requeue=True):
                content = json.loads(message.body.decode())
                await handler(content)

        await queue.consume(on_message)

    async def close(self) -> None:
        if self.connection:
            await self.connection.close()


# Usage
async def main():
    mq = MessageQueue()
    await mq.connect()

    await mq.publish_event("user.created", {"userId": 123, "email": "user@example.com"})

    async def handle_user_created(event: dict):
        print("New user created:", event["payload"])

    await mq.subscribe_to_event("user.created", handle_user_created)

using MassTransit;
using Microsoft.Extensions.DependencyInjection;
using System.Text.Json;

// Producer service
public class MessageQueueService
{
    private readonly IPublishEndpoint _publishEndpoint;

    public MessageQueueService(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }

    public async Task PublishEventAsync(string eventType, object payload)
    {
        var message = new EventMessage
        {
            Type = eventType,
            Timestamp = DateTime.UtcNow,
            Payload = JsonSerializer.Serialize(payload)
        };

        await _publishEndpoint.Publish(message);
        Console.WriteLine($"Published event: {eventType}");
    }
}

public record EventMessage
{
    public string Type { get; init; } = "";
    public DateTime Timestamp { get; init; }
    public string Payload { get; init; } = "";
}

// Consumer: implement IConsumer<T> with MassTransit
public class UserCreatedConsumer : IConsumer<EventMessage>
{
    public async Task Consume(ConsumeContext<EventMessage> context)
    {
        var payload = JsonSerializer.Deserialize<Dictionary<string, object>>(
            context.Message.Payload);
        Console.WriteLine($"New user created: {payload}");
        // Send welcome email, initialize user profile, etc.
        await Task.CompletedTask;
    }
}

// Registration (Program.cs / Startup)
// services.AddMassTransit(x => {
//     x.AddConsumer<UserCreatedConsumer>();
//     x.UsingRabbitMq((ctx, cfg) => {
//         cfg.Host("localhost");
//         cfg.ConfigureEndpoints(ctx);
//     });
// });

// Token bucket algorithm
class TokenBucket {
  private tokens: number;
  private refillRate: number; // tokens per second
  private lastRefillTime: number = Date.now();
  private maxTokens: number;

  constructor(capacity: number, refillRate: number) {
    this.maxTokens = capacity;
    this.tokens = capacity;
    this.refillRate = refillRate;
  }

  private refill(): void {
    const now = Date.now();
    const timePassed = (now - this.lastRefillTime) / 1000;
    const tokensToAdd = timePassed * this.refillRate;

    this.tokens = Math.min(this.maxTokens, this.tokens + tokensToAdd);
    this.lastRefillTime = now;
  }

  allowRequest(tokensRequired: number = 1): boolean {
    this.refill();

    if (this.tokens >= tokensRequired) {
      this.tokens -= tokensRequired;
      return true;
    }

    return false;
  }
}

// Per-user rate limiting middleware
const userBuckets = new Map<string, TokenBucket>();

function rateLimitMiddleware(req: any, res: any, next: any) {
  const userId = req.user.id;
  
  // 100 requests per minute
  if (!userBuckets.has(userId)) {
    userBuckets.set(userId, new TokenBucket(100, 100 / 60));
  }

  const bucket = userBuckets.get(userId)!;

  if (!bucket.allowRequest()) {
    return res.status(429).json({
      error: 'Rate limit exceeded',
      retryAfter: 60,
    });
  }

  res.setHeader('X-RateLimit-Remaining', bucket.tokens);
  next();
}

import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
import jakarta.servlet.*;
import jakarta.servlet.http.*;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.*;

@Component
public class RateLimitFilter extends OncePerRequestFilter {

    private final ConcurrentHashMap<String, TokenBucket> userBuckets = new ConcurrentHashMap<>();

    @Override
    protected void doFilterInternal(HttpServletRequest request,
                                     HttpServletResponse response,
                                     FilterChain chain) throws IOException, ServletException {
        String userId = (String) request.getAttribute("userId");
        if (userId == null) { chain.doFilter(request, response); return; }

        TokenBucket bucket = userBuckets.computeIfAbsent(userId,
            k -> new TokenBucket(100, 100.0 / 60.0)); // 100 req/min

        if (!bucket.allowRequest()) {
            response.setStatus(429);
            response.setContentType("application/json");
            response.getWriter().write("{\"error\":\"Rate limit exceeded\",\"retryAfter\":60}");
            return;
        }

        response.setHeader("X-RateLimit-Remaining", String.valueOf((int) bucket.getTokens()));
        chain.doFilter(request, response);
    }

    static class TokenBucket {
        private double tokens;
        private final double refillRate;
        private long lastRefillTime;
        private final double maxTokens;

        TokenBucket(double capacity, double refillRate) {
            this.maxTokens = capacity;
            this.tokens = capacity;
            this.refillRate = refillRate;
            this.lastRefillTime = System.currentTimeMillis();
        }

        private synchronized void refill() {
            long now = System.currentTimeMillis();
            double timePassed = (now - lastRefillTime) / 1000.0;
            tokens = Math.min(maxTokens, tokens + timePassed * refillRate);
            lastRefillTime = now;
        }

        public synchronized boolean allowRequest() {
            refill();
            if (tokens >= 1) { tokens--; return true; }
            return false;
        }

        public synchronized double getTokens() { return tokens; }
    }
}

import time
import threading
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()


class TokenBucket:
    def __init__(self, capacity: float, refill_rate: float):
        self.max_tokens = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate
        self.last_refill = time.time()
        self._lock = threading.Lock()

    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.max_tokens, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

    def allow_request(self, tokens_required: float = 1.0) -> bool:
        with self._lock:
            self._refill()
            if self.tokens >= tokens_required:
                self.tokens -= tokens_required
                return True
            return False


user_buckets: dict[str, TokenBucket] = {}
_buckets_lock = threading.Lock()


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    user_id = getattr(request.state, "user_id", None)
    if user_id is None:
        return await call_next(request)

    with _buckets_lock:
        if user_id not in user_buckets:
            user_buckets[user_id] = TokenBucket(100, 100 / 60)  # 100 req/min

    bucket = user_buckets[user_id]

    if not bucket.allow_request():
        return JSONResponse(
            status_code=429,
            content={"error": "Rate limit exceeded", "retryAfter": 60},
        )

    response = await call_next(request)
    response.headers["X-RateLimit-Remaining"] = str(int(bucket.tokens))
    return response

using Microsoft.AspNetCore.Http;
using System.Collections.Concurrent;
using System.Text.Json;

public class RateLimitMiddleware
{
    private readonly RequestDelegate _next;
    private readonly ConcurrentDictionary<string, TokenBucket> _userBuckets = new();

    public RateLimitMiddleware(RequestDelegate next) => _next = next;

    public async Task InvokeAsync(HttpContext context)
    {
        var userId = context.User?.Identity?.Name;
        if (userId is null) { await _next(context); return; }

        var bucket = _userBuckets.GetOrAdd(userId,
            _ => new TokenBucket(100, 100.0 / 60.0)); // 100 req/min

        if (!bucket.AllowRequest())
        {
            context.Response.StatusCode = 429;
            context.Response.ContentType = "application/json";
            await context.Response.WriteAsync(
                JsonSerializer.Serialize(new { error = "Rate limit exceeded", retryAfter = 60 }));
            return;
        }

        context.Response.Headers["X-RateLimit-Remaining"] = ((int)bucket.Tokens).ToString();
        await _next(context);
    }
}

public class TokenBucket
{
    private double _tokens;
    private readonly double _refillRate;
    private long _lastRefillMs;
    private readonly double _maxTokens;
    private readonly object _lock = new();

    public TokenBucket(double capacity, double refillRate)
    {
        _maxTokens = capacity;
        _tokens = capacity;
        _refillRate = refillRate;
        _lastRefillMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
    }

    private void Refill()
    {
        var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
        var elapsed = (now - _lastRefillMs) / 1000.0;
        _tokens = Math.Min(_maxTokens, _tokens + elapsed * _refillRate);
        _lastRefillMs = now;
    }

    public bool AllowRequest(double tokensRequired = 1)
    {
        lock (_lock)
        {
            Refill();
            if (_tokens >= tokensRequired) { _tokens -= tokensRequired; return true; }
            return false;
        }
    }

    public double Tokens { get { lock (_lock) { Refill(); return _tokens; } } }
}

enum CircuitState {
  CLOSED = 'CLOSED',
  OPEN = 'OPEN',
  HALF_OPEN = 'HALF_OPEN',
}

class CircuitBreaker {
  private state: CircuitState = CircuitState.CLOSED;
  private failureCount: number = 0;
  private successCount: number = 0;
  private lastFailureTime: number = 0;
  private readonly failureThreshold: number = 5;
  private readonly successThreshold: number = 2;
  private readonly timeout: number = 60000; // 60 seconds

  async execute<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === CircuitState.OPEN) {
      if (Date.now() - this.lastFailureTime > this.timeout) {
        this.state = CircuitState.HALF_OPEN;
        this.successCount = 0;
      } else {
        throw new Error('Circuit breaker is OPEN');
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  private onSuccess(): void {
    this.failureCount = 0;

    if (this.state === CircuitState.HALF_OPEN) {
      this.successCount++;
      if (this.successCount >= this.successThreshold) {
        this.state = CircuitState.CLOSED;
      }
    }
  }

  private onFailure(): void {
    this.lastFailureTime = Date.now();
    this.failureCount++;

    if (this.failureCount >= this.failureThreshold) {
      this.state = CircuitState.OPEN;
    }
  }

  getState(): CircuitState {
    return this.state;
  }
}

// Usage
const breaker = new CircuitBreaker();

async function callExternalAPI(): Promise<any> {
  return breaker.execute(() =>
    fetch('https://external-api.com/data').then(r => r.json())
  );
}