Recipe Gallery

def add_score_with_tiebreak(game_id: str, player_id: str, points: int) -> None:
    """Score = points + fractional inverse-timestamp.
    Example: 1500 pts at epoch 1700000000 -> 1500.300000000
    Higher fractional = earlier timestamp, so ZREVRANGE ranks earlier achievers first.
    """
    max_ts = 9999999999  # far-future ceiling
    fractional = (max_ts - int(time.time())) / (max_ts + 1)
    get_client().zadd(f"leaderboard:{game_id}", {player_id: points + fractional})

def get_top_n(game_id: str, n: int = 10) -> list[tuple[str, float]]:
    return get_client().zrevrange(f"leaderboard:{game_id}", 0, n - 1, withscores=True)

def get_player_rank(game_id: str, player_id: str) -> int | None:
    return get_client().zrevrank(f"leaderboard:{game_id}", player_id)

CACHE_TTL = 300  # 5 minutes

def cache_aside_get(entity: str, entity_id: str, db_fetch_fn, ttl: int = CACHE_TTL) -> dict:
    """Cache-aside with XFetch. db_fetch_fn(entity_id) -> dict reads from the DB."""
    key = f"cache:{entity}:{entity_id}"
    c = get_client()
    cached = c.get(key)
    if cached is not None:
        data = json.loads(cached)
        remaining_ttl = c.ttl(key)
        if remaining_ttl > 0 and _should_early_refresh(remaining_ttl, ttl):
            result = db_fetch_fn(entity_id)
            if result is not None:
                c.setex(key, ttl, json.dumps(result))
                return result
        return data
    # Cache miss
    result = db_fetch_fn(entity_id)
    if result is not None:
        c.setex(key, ttl, json.dumps(result))
    return result

def _should_early_refresh(remaining_ttl: int, total_ttl: int) -> bool:
    """XFetch: returns True with increasing probability as TTL approaches 0.
    Only activates in the last 20% of TTL. beta controls aggressiveness."""
    beta = 1.0
    if total_ttl <= 0:
        return False
    threshold = total_ttl * 0.2
    if remaining_ttl >= threshold:
        return False
    return random.random() < math.exp(-remaining_ttl * beta / threshold)

def sliding_window_rate_limit(
    client_id: str, window_name: str, max_requests: int, window_seconds: int,
) -> tuple[bool, int]:
    """Returns (allowed, remaining)."""
    key = f"ratelimit:{client_id}:{window_name}"
    now = time.time()
    member = f"{now}:{uuid.uuid4().hex[:8]}"
    c = get_client()

    pipe = c.pipeline()
    pipe.zremrangebyscore(key, "-inf", now - window_seconds)  # prune old
    pipe.zadd(key, {member: now})                             # optimistic add
    pipe.zcard(key)                                           # count window
    pipe.expire(key, window_seconds + 1)                      # auto-cleanup
    results = pipe.execute()

    current_count = results[2]
    allowed = current_count <= max_requests
    if not allowed:
        c.zrem(key, member)  # remove denied request to keep count accurate
    return allowed, max(0, max_requests - current_count)

CACHE_TTL = 300  # 5-minute backstop

def write_through(entity: str, entity_id: str, data: dict, db_write_fn) -> None:
    """Write to DB then cache. TTL acts as safety net."""
    db_write_fn(entity_id, data)
    get_client().setex(f"cache:{entity}:{entity_id}", CACHE_TTL, json.dumps(data))

def handle_change_event(entity: str, entity_id: str) -> None:
    """Called by SNS/SQS consumer when another service modifies the entity."""
    get_client().delete(f"cache:{entity}:{entity_id}")