System Design in Coding Questions

import time
from collections import defaultdict

BUCKET_SIZE = 60  # seconds

def bucket_ts(ts, size=BUCKET_SIZE):
    return ts - (ts % size)

class CounterAggregator:
    def __init__(self, bucket_size=BUCKET_SIZE):
        self.bucket_size = bucket_size
        self.buckets = defaultdict(int)  # key: (metric, bucket_ts) -> sum

    def ingest(self, metric_name, tags, value, ts=None):
        ts = ts or int(time.time())
        b = bucket_ts(ts, self.bucket_size)
        key = (metric_name, tuple(sorted(tags.items())))
        self.buckets[(key, b)] += int(value)

    def flush_bucket(self, b):
        results = []
        to_delete = []
        for (key, bucket_time), total in list(self.buckets.items()):
            if bucket_time == b:
                metric_name, tags_tuple = key
                tags = dict(tags_tuple)
                results.append({
                    "metric": metric_name,
                    "tags": tags,
                    "timestamp": bucket_time,
                    "value": total
                })
                to_delete.append((key, bucket_time))
        for k in to_delete:
            del self.buckets[k]
        return results

# Usage: ingest continuously; every BUCKET_SIZE seconds call flush_bucket(current_bucket-1)

# python
from collections import defaultdict

class ID:
    def __init__(self, replica, counter):
        self.replica = replica
        self.counter = counter
    def __lt__(self, other):
        return (self.counter, self.replica) < (other.counter, other.replica)
    def __repr__(self):
        return f"({self.replica},{self.counter})"

class RGA:
    def __init__(self):
        self.nodes = {}           # id -> (char, visible, nexts list)
        self.nexts = defaultdict(list)
        self.head = ID("",0)
        self.nodes[self.head] = (None, True)
    def insert(self, prev_id, char, id):
        self.nodes[id] = (char, True)
        self.nexts[prev_id].append(id)
        # deterministic order
        self.nexts[prev_id].sort()
    def tombstone(self, id):
        char, _ = self.nodes[id]
        self.nodes[id] = (char, False)
    def to_str(self):
        res=[]
        def walk(cur):
            for nid in self.nexts[cur]:
                ch, vis = self.nodes[nid]
                if vis: res.append(ch)
                walk(nid)
        walk(self.head)
        return "".join(res)

# Simulate concurrent inserts
a = RGA(); b = RGA()
# both start from head
id_a1 = ID("A",1); id_b1 = ID("B",1)
a.insert(a.head, "x", id_a1)   # A inserts 'x' after head
b.insert(b.head, "y", id_b1)   # B inserts 'y' after head

# Merge ops into a
a.insert(a.head, "y", id_b1)
# Merge ops into b
b.insert(b.head, "x", id_a1)

print(a.to_str(), b.to_str())  # both deterministically "xy" or "yx" depending on ID ordering

# python
import time
import redis

r = redis.Redis()

# Lua script for atomic token bucket
TOKEN_BUCKET_LUA = """
local key, rate, burst, now, cost = KEYS[1], tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3]), tonumber(ARGV[4])
local state = redis.call('HMGET', key, 'tokens', 'ts')
local tokens = tonumber(state[1]) or burst
local ts = tonumber(state[2]) or now
local dt = math.max(0, now - ts)
tokens = math.min(burst, tokens + dt * rate)
if tokens < cost then
  redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
  return 0
else
  tokens = tokens - cost
  redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
  return 1
end
"""

token_bucket = r.register_script(TOKEN_BUCKET_LUA)

def allow_request(tenant_id, rate_per_sec, burst, cost=1):
    key = f"tb:{tenant_id}"
    now = time.time()
    allowed = token_bucket(keys=[key], args=[str(rate_per_sec), str(burst), str(now), str(cost)])
    return bool(int(allowed))

# Usage in middleware
def middleware(request):
    tenant = resolve_tenant(request)
    # rates fetched from config/cache per-tenant
    rate, burst = get_tenant_rate(tenant)
    if not allow_request(tenant, rate, burst):
        return http_429_response()
    return forward_request(request)

# publisher.py (pseudo)
import psycopg2, json, time
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='kafka:9092', value_serializer=lambda v: json.dumps(v).encode())

def write_entity(conn, entity_id, new_data, new_version):
    # Begin transaction
    cur = conn.cursor()
    cur.execute("BEGIN;")
    # Update source-of-truth
    cur.execute("UPDATE entities SET data = %s, version = %s WHERE id = %s;", (json.dumps(new_data), new_version, entity_id))
    # Insert outbox event in same tx
    event = {"entity_id": entity_id, "version": new_version, "op": "update", "ts": int(time.time()*1000)}
    cur.execute("INSERT INTO outbox (aggregate_id, topic, payload, published) VALUES (%s, %s, %s, false);", (entity_id, "entity-updates", json.dumps(event)))
    cur.execute("COMMIT;")

# relay.py - reliably publish outbox rows to Kafka (idempotent)
import time, json
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='kafka:9092', value_serializer=lambda v: json.dumps(v).encode(), acks='all', enable_idempotence=True)

def publish_outbox(conn):
    cur = conn.cursor()
    # select un-published rows
    cur.execute("SELECT id, aggregate_id, payload FROM outbox WHERE published = false ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 100;")
    rows = cur.fetchall()
    for r in rows:
        outbox_id, key, payload = r
        producer.send("entity-updates", key=str(key).encode(), value=json.loads(payload))
        producer.flush()
        # mark published
        cur.execute("UPDATE outbox SET published = true WHERE id = %s;", (outbox_id,))
    conn.commit()

# run periodically / in loop

# consumer.py - invalidate local cache safely
from kafka import KafkaConsumer
consumer = KafkaConsumer('entity-updates', bootstrap_servers='kafka:9092', value_deserializer=lambda v: json.loads(v.decode()), group_id='service-A', auto_offset_reset='earliest')

local_cache = {}  # entity_id -> (version, value)

for msg in consumer:
    event = msg.value
    eid = event['entity_id']
    ver = event['version']
    # Only apply if newer than cached version
    cur_ver, _ = local_cache.get(eid, (0, None))
    if ver > cur_ver:
        # Evict or update cached value. Option: lazy fetch authoritative data or accept event payload
        local_cache[eid] = (ver, None)  # mark invalidated; next read will fetch from source-of-truth
    # commit offset automatically or explicitly after processing