Mini Redis 직접 구현하기: Geo 캐시까지 꿰뚫어 보기

// 100만 개 삽입 후 10초 대기
repeat(1_000_000) { i ->
    store.set("key-$i", "value", ttlMs = 1000)
}
Thread.sleep(10_000)
println(store.size())  // 출력: 1000000 (삭제되지 않음!)

// Duration으로 받으면 유연함
store.set("key", "value", ttl = Duration.ofMillis(500))
store.set("key", "value", ttl = Duration.ofHours(1))
store.set("key", "value", ttl = Duration.ofDays(7))

enum class DataType {
    STRING, LIST, SET, SORTED_SET, HASH
}

data class RedisValue(
    val data: Any,
    val type: DataType,
    val expireAt: Long? = null
) {
    fun isExpired(): Boolean =
        expireAt != null && System.currentTimeMillis() > expireAt
}

class MiniRedis {
    private val store = ConcurrentHashMap<String, RedisValue>()
    private val ttlManager = TTLManager(store)

    init {
        ttlManager.startActiveExpiration()
    }

    fun set(key: String, value: Any, ttlMs: Long? = null) {
        val expireAt = ttlMs?.let { System.currentTimeMillis() + it }
        store[key] = RedisValue(value, DataType.STRING, expireAt)
    }

    fun get(key: String): Any? {
        val value = store[key] ?: return null
        // Lazy expiration
        if (value.isExpired()) {
            store.remove(key)
            return null
        }
        return value.data
    }

    fun delete(key: String): Boolean = store.remove(key) != null

    fun size(): Int = store.size

    fun keys(): Set<String> = store.keys.toSet()
}

class TTLManager(private val store: ConcurrentHashMap<String, RedisValue>) {
    private val sampleSize = 20
    private val expiredThreshold = 0.25

    fun startActiveExpiration(intervalMs: Long = 100) {
        thread(isDaemon = true, name = "ttl-manager") {
            while (true) {
                Thread.sleep(intervalMs)
                expireSample()
            }
        }
    }

    private fun expireSample() {
        if (store.isEmpty()) return

        var expired = 0
        val keys = store.keys().toList()
        val sample = keys.shuffled().take(minOf(sampleSize, keys.size))

        for (key in sample) {
            val value = store[key] ?: continue
            if (value.isExpired()) {
                store.remove(key)
                expired++
            }
        }

        // 만료된 키가 25% 이상이면 다시 샘플링
        if (expired > sample.size * expiredThreshold) {
            expireSample()
        }
    }
}

// LinkedList 기반 구현
class RedisList {
    private val list = LinkedList<String>()

    fun lpush(value: String) = list.addFirst(value)  // O(1)
    fun rpop(): String? = list.pollLast()            // O(1)
    fun lindex(index: Int): String? = list.getOrNull(index)  // O(N)!
}

class RedisSet {
    private val set = mutableSetOf<String>()

    fun sadd(value: String): Boolean = set.add(value)
    fun srem(value: String): Boolean = set.remove(value)
}

data class ScoreEntry(val member: String, val score: Double) : Comparable<ScoreEntry> {
    override fun compareTo(other: ScoreEntry): Int {
        val scoreCompare = score.compareTo(other.score)
        return if (scoreCompare != 0) scoreCompare else member.compareTo(other.member)
    }
}

class SortedSet {
    private val data = ConcurrentSkipListSet<ScoreEntry>()

    fun zadd(member: String, score: Double) {
        data.removeIf { it.member == member }  // 기존 점수 제거
        data.add(ScoreEntry(member, score))
    }

    fun zrange(start: Int, stop: Int): List<String> {
        return data.drop(start).take(stop - start + 1).map { it.member }
    }

    fun zrangeByScore(min: Double, max: Double): List<String> {
        return data.filter { it.score in min..max }.map { it.member }
    }
}

// List 구현
class RedisList {
    private val list = Collections.synchronizedList(LinkedList<String>())

    fun lpush(vararg values: String): Int {
        values.forEach { list.add(0, it) }
        return list.size
    }

    fun rpush(vararg values: String): Int {
        values.forEach { list.add(it) }
        return list.size
    }

    fun lpop(): String? = if (list.isNotEmpty()) list.removeAt(0) else null

    fun rpop(): String? = if (list.isNotEmpty()) list.removeAt(list.size - 1) else null

    fun lrange(start: Int, stop: Int): List<String> {
        val end = if (stop < 0) list.size + stop + 1 else minOf(stop + 1, list.size)
        val actualStart = if (start < 0) maxOf(0, list.size + start) else start
        return if (actualStart < end) list.subList(actualStart, end).toList() else emptyList()
    }

    fun llen(): Int = list.size
}

// Set 구현
class RedisSet {
    private val set = ConcurrentHashMap.newKeySet<String>()

    fun sadd(vararg members: String): Int {
        var added = 0
        members.forEach { if (set.add(it)) added++ }
        return added
    }

    fun srem(vararg members: String): Int {
        var removed = 0
        members.forEach { if (set.remove(it)) removed++ }
        return removed
    }

    fun sismember(member: String): Boolean = set.contains(member)

    fun smembers(): Set<String> = set.toSet()

    fun scard(): Int = set.size
}

// Sorted Set 구현
data class ScoreEntry(val member: String, val score: Double) : Comparable<ScoreEntry> {
    override fun compareTo(other: ScoreEntry): Int {
        val scoreCompare = score.compareTo(other.score)
        return if (scoreCompare != 0) scoreCompare else member.compareTo(other.member)
    }
}

class RedisSortedSet {
    private val data = ConcurrentSkipListSet<ScoreEntry>()
    private val memberScores = ConcurrentHashMap<String, Double>()

    fun zadd(member: String, score: Double): Int {
        val existing = memberScores[member]
        if (existing != null) {
            data.remove(ScoreEntry(member, existing))
        }
        data.add(ScoreEntry(member, score))
        memberScores[member] = score
        return if (existing == null) 1 else 0
    }

    fun zrem(member: String): Int {
        val score = memberScores.remove(member) ?: return 0
        data.remove(ScoreEntry(member, score))
        return 1
    }

    fun zscore(member: String): Double? = memberScores[member]

    fun zrange(start: Int, stop: Int): List<String> {
        val end = if (stop < 0) data.size + stop + 1 else minOf(stop + 1, data.size)
        return data.drop(start).take(end - start).map { it.member }
    }

    fun zrangeByScore(min: Double, max: Double): List<String> {
        return data.filter { it.score in min..max }.map { it.member }
    }

    fun zcard(): Int = data.size
}

private val channels = ConcurrentHashMap<String, MutableList<(String) -> Unit>>()

fun publish(channel: String, message: String) {
    channels[channel]?.forEach { handler ->
        handler(message)  // 순회 중 subscribe/unsubscribe가 발생하면?
    }
}

fun publish(channel: String, message: String) {
    channels[channel]?.forEach { handler ->
        handler(message)  // 이 핸들러가 1초 걸리면 다음 핸들러도 1초 대기
    }
}

class PubSub {
    private val channels = ConcurrentHashMap<String, CopyOnWriteArrayList<Subscriber>>()

    data class Subscriber(
        val id: String,
        val handler: (String) -> Unit
    )

    fun subscribe(channel: String, subscriberId: String, handler: (String) -> Unit) {
        channels.computeIfAbsent(channel) { CopyOnWriteArrayList() }
            .add(Subscriber(subscriberId, handler))
    }

    fun unsubscribe(channel: String, subscriberId: String) {
        val subscribers = channels[channel] ?: return
        subscribers.removeIf { it.id == subscriberId }

        // 구독자가 0명이면 채널 삭제
        if (subscribers.isEmpty()) {
            channels.remove(channel)
        }
    }

    fun publish(channel: String, message: String): Int {
        val subscribers = channels[channel] ?: return 0
        subscribers.forEach { it.handler(message) }
        return subscribers.size
    }

    fun getSubscriberCount(channel: String): Int {
        return channels[channel]?.size ?: 0
    }

    fun listChannels(): Set<String> = channels.keys.toSet()
}

// RDB 스냅샷
class RdbPersister(private val store: ConcurrentHashMap<String, RedisValue>) {
    fun save(path: Path) {
        ObjectOutputStream(Files.newOutputStream(path)).use { out ->
            out.writeObject(store.toMap())
        }
    }

    @Suppress("UNCHECKED_CAST")
    fun load(path: Path) {
        if (!Files.exists(path)) return

        ObjectInputStream(Files.newInputStream(path)).use { input ->
            val data = input.readObject() as Map<String, RedisValue>
            store.clear()
            store.putAll(data)
        }
    }
}

// AOF 로그
class AofPersister(private val path: Path) {
    private val writer = Files.newBufferedWriter(
        path,
        StandardOpenOption.CREATE,
        StandardOpenOption.APPEND
    )

    @Synchronized
    fun log(command: String, args: List<String>) {
        val line = buildString {
            append(command)
            args.forEach { append(" ").append(escape(it)) }
            append("\n")
        }
        writer.write(line)
        writer.flush()
    }

    fun replay(store: MiniRedis) {
        if (!Files.exists(path)) return

        Files.readAllLines(path).forEach { line ->
            val parts = parseLine(line)
            if (parts.isNotEmpty()) {
                executeCommand(store, parts[0], parts.drop(1))
            }
        }
    }

    fun rewrite(store: ConcurrentHashMap<String, RedisValue>) {
        val tempPath = path.resolveSibling("temp_${path.fileName}")
        Files.newBufferedWriter(tempPath).use { writer ->
            store.forEach { (key, value) ->
                val command = when (value.type) {
                    DataType.STRING -> "SET $key ${escape(value.data.toString())}"
                    else -> return@forEach  // 다른 타입은 생략
                }
                val expireCmd = value.expireAt?.let { " EXPIREAT $key $it" } ?: ""
                writer.write("$command$expireCmd\n")
            }
        }
        Files.move(tempPath, path, StandardCopyOption.REPLACE_EXISTING)
    }

    private fun escape(s: String): String = "\"${s.replace("\"", "\\\"")}\""
    private fun parseLine(line: String): List<String> = line.split(" ")  // 간소화
    private fun executeCommand(store: MiniRedis, cmd: String, args: List<String>) {
        when (cmd.uppercase()) {
            "SET" -> if (args.size >= 2) store.set(args[0], args[1])
            "DEL" -> if (args.isNotEmpty()) store.delete(args[0])
        }
    }
}

class ExactLRU<K, V>(private val maxSize: Int) {
    private val map = LinkedHashMap<K, V>(maxSize, 0.75f, true)  // accessOrder=true

    fun get(key: K): V? = map[key]

    fun put(key: K, value: V) {
        map[key] = value
        if (map.size > maxSize) {
            val eldest = map.entries.first()
            map.remove(eldest.key)
        }
    }
}

fun evictLRUSample(sampleSize: Int = 5) {
    val sample = store.entries.shuffled().take(sampleSize)
    val oldest = sample.minByOrNull { it.value.lastAccessTime }
    oldest?.let { store.remove(it.key) }
}

enum class EvictionPolicy {
    LRU_SAMPLING,   // 샘플링 기반 LRU
    TTL_FIRST,      // TTL 만료 임박 우선
    RANDOM          // 랜덤 삭제
}

class EvictionManager(
    private val store: ConcurrentHashMap<String, RedisValue>,
    private val policy: EvictionPolicy = EvictionPolicy.LRU_SAMPLING,
    private val maxMemoryBytes: Long = 100 * 1024 * 1024,  // 100MB
    private val sampleSize: Int = 5
) {
    private val accessTimes = ConcurrentHashMap<String, Long>()

    fun recordAccess(key: String) {
        accessTimes[key] = System.currentTimeMillis()
    }

    fun checkAndEvict() {
        val currentMemory = estimateMemory()
        if (currentMemory < maxMemoryBytes * 0.8) return

        val keysToEvict = mutableListOf<String>()

        // 메모리가 70% 이하가 될 때까지 eviction
        while (estimateMemory() > maxMemoryBytes * 0.7 && store.isNotEmpty()) {
            val keyToEvict = selectKeyToEvict() ?: break
            keysToEvict.add(keyToEvict)
            store.remove(keyToEvict)
            accessTimes.remove(keyToEvict)
        }

        if (keysToEvict.isNotEmpty()) {
            println("Evicted ${keysToEvict.size} keys")
        }
    }

    private fun selectKeyToEvict(): String? {
        return when (policy) {
            EvictionPolicy.LRU_SAMPLING -> selectLRUSample()
            EvictionPolicy.TTL_FIRST -> selectTTLFirst()
            EvictionPolicy.RANDOM -> selectRandom()
        }
    }

    private fun selectLRUSample(): String? {
        val sample = store.keys.toList().shuffled().take(sampleSize)
        return sample.minByOrNull { accessTimes[it] ?: 0L }
    }

    private fun selectTTLFirst(): String? {
        // TTL이 있고 만료가 가까운 키 우선
        val withTTL = store.entries.filter { it.value.expireAt != null }
        if (withTTL.isNotEmpty()) {
            return withTTL.minByOrNull { it.value.expireAt!! }?.key
        }
        // TTL이 없으면 LRU
        return selectLRUSample()
    }

    private fun selectRandom(): String? {
        return store.keys.randomOrNull()
    }

    private fun estimateMemory(): Long {
        // 간단한 추정: 키당 평균 100바이트로 계산
        return store.size * 100L
    }
}

// 위도 37.5665, 경도 126.9780의 Geohash 계산 과정
// 1. 경도 범위 [-180, 180]을 이진 검색으로 좁혀간다
// 2. 위도 범위 [-90, 90]을 이진 검색으로 좁혀간다
// 3. 경도/위도 비트를 번갈아 배치
// 4. 5비트씩 묶어서 base32로 변환
// 결과: "wydm9q..."

// Geohash를 정수로 변환해서 score로 사용
val scoreMin = geohashToLong("wydm9q")  // 범위 시작
val scoreMax = geohashToLong("wydm9r")  // 범위 끝 (다음 접두사)
sortedSet.zrangeByScore(scoreMin, scoreMax)  // 해당 구역의 모든 카페

data class GeoCoordinate(val latitude: Double, val longitude: Double)

class GeoHash {
    private val base32 = "0123456789bcdefghjkmnpqrstuvwxyz"

    fun encode(lat: Double, lon: Double, precision: Int = 6): String {
        var latMin = -90.0
        var latMax = 90.0
        var lonMin = -180.0
        var lonMax = 180.0

        var bits = 0
        var bitsTotal = 0
        var hashValue = 0
        val result = StringBuilder()

        while (result.length < precision) {
            if (bitsTotal % 2 == 0) {
                // 경도
                val mid = (lonMin + lonMax) / 2
                if (lon >= mid) {
                    hashValue = (hashValue shl 1) + 1
                    lonMin = mid
                } else {
                    hashValue = hashValue shl 1
                    lonMax = mid
                }
            } else {
                // 위도
                val mid = (latMin + latMax) / 2
                if (lat >= mid) {
                    hashValue = (hashValue shl 1) + 1
                    latMin = mid
                } else {
                    hashValue = hashValue shl 1
                    latMax = mid
                }
            }

            bitsTotal++
            bits++

            if (bits == 5) {
                result.append(base32[hashValue])
                bits = 0
                hashValue = 0
            }
        }

        return result.toString()
    }

    fun decode(hash: String): Pair<ClosedRange<Double>, ClosedRange<Double>> {
        var latMin = -90.0
        var latMax = 90.0
        var lonMin = -180.0
        var lonMax = 180.0
        var isLon = true

        for (c in hash) {
            val bits = base32.indexOf(c)
            for (i in 4 downTo 0) {
                val bit = (bits shr i) and 1
                if (isLon) {
                    val mid = (lonMin + lonMax) / 2
                    if (bit == 1) lonMin = mid else lonMax = mid
                } else {
                    val mid = (latMin + latMax) / 2
                    if (bit == 1) latMin = mid else latMax = mid
                }
                isLon = !isLon
            }
        }

        return (latMin..latMax) to (lonMin..lonMax)
    }

    fun neighbors(hash: String): List<String> {
        // 인접한 8방향의 Geohash 반환 (구현 생략)
        // N, NE, E, SE, S, SW, W, NW
        return emptyList()
    }
}

class GeoCache(private val precision: Int = 6) {
    private val cache = ConcurrentHashMap<String, MutableList<GeoEntry>>()
    private val geoHash = GeoHash()

    data class GeoEntry(
        val id: String,
        val coordinate: GeoCoordinate,
        val data: Any
    )

    fun add(id: String, lat: Double, lon: Double, data: Any) {
        val hash = geoHash.encode(lat, lon, precision)
        cache.computeIfAbsent(hash) { mutableListOf() }
            .add(GeoEntry(id, GeoCoordinate(lat, lon), data))
    }

    fun searchNearby(lat: Double, lon: Double): List<GeoEntry> {
        val centerHash = geoHash.encode(lat, lon, precision)
        val hashes = listOf(centerHash) + geoHash.neighbors(centerHash)

        return hashes.flatMap { hash ->
            cache[hash] ?: emptyList()
        }
    }

    fun searchInRadius(lat: Double, lon: Double, radiusKm: Double): List<GeoEntry> {
        return searchNearby(lat, lon).filter { entry ->
            haversineDistance(lat, lon, entry.coordinate.latitude, entry.coordinate.longitude) <= radiusKm
        }
    }

    private fun haversineDistance(lat1: Double, lon1: Double, lat2: Double, lon2: Double): Double {
        val R = 6371.0  // 지구 반지름 (km)
        val dLat = Math.toRadians(lat2 - lat1)
        val dLon = Math.toRadians(lon2 - lon1)
        val a = sin(dLat / 2).pow(2) + cos(Math.toRadians(lat1)) * cos(Math.toRadians(lat2)) * sin(dLon / 2).pow(2)
        val c = 2 * atan2(sqrt(a), sqrt(1 - a))
        return R * c
    }
}

해시 링
     0
   /   \
Node1   Node2
   \   /
  Node3

키 "cafe:123"의 해시값이 Node2와 Node3 사이에 있으면 Node3에 저장

// 물리 노드 1개당 가상 노드 150개
val virtualNodesPerNode = 150

for (node in physicalNodes) {
    for (i in 0 until virtualNodesPerNode) {
        val virtualNodeHash = hash("${node.id}-$i")
        ring[virtualNodeHash] = node
    }
}

fun getNodesForKey(key: String): List<Node> {
    val primaryNode = ring.getNodeForKey(key)
    val replicaNode = ring.getNextNode(primaryNode)
    return listOf(primaryNode, replicaNode)
}

class ConsistentHashRing(
    private val virtualNodesPerNode: Int = 150
) {
    private val ring = TreeMap<Long, Node>()
    private val nodes = mutableListOf<Node>()

    data class Node(
        val id: String,
        val host: String,
        val port: Int
    )

    fun addNode(node: Node) {
        nodes.add(node)
        for (i in 0 until virtualNodesPerNode) {
            val hash = hash("${node.id}-$i")
            ring[hash] = node
        }
    }

    fun removeNode(node: Node) {
        nodes.remove(node)
        for (i in 0 until virtualNodesPerNode) {
            val hash = hash("${node.id}-$i")
            ring.remove(hash)
        }
    }

    fun getNodeForKey(key: String): Node? {
        if (ring.isEmpty()) return null
        val hash = hash(key)
        val tailMap = ring.tailMap(hash)
        val nodeHash = if (tailMap.isEmpty()) ring.firstKey() else tailMap.firstKey()
        return ring[nodeHash]
    }

    fun getNodesForKey(key: String, replicaCount: Int = 2): List<Node> {
        if (ring.isEmpty()) return emptyList()

        val result = mutableListOf<Node>()
        val hash = hash(key)
        var currentHash = hash

        while (result.size < replicaCount && result.size < nodes.size) {
            val tailMap = ring.tailMap(currentHash)
            val nodeHash = if (tailMap.isEmpty()) ring.firstKey() else tailMap.firstKey()
            val node = ring[nodeHash]!!

            if (node !in result) {
                result.add(node)
            }
            currentHash = nodeHash + 1
        }

        return result
    }

    private fun hash(key: String): Long {
        val md = MessageDigest.getInstance("MD5")
        val digest = md.digest(key.toByteArray())
        return ByteBuffer.wrap(digest.take(8).toByteArray()).long
    }
}

class ShardedCache(
    private val ring: ConsistentHashRing,
    private val replicaCount: Int = 2
) {
    private val nodeStores = ConcurrentHashMap<String, MiniRedis>()

    fun set(key: String, value: Any, ttlMs: Long? = null) {
        val nodes = ring.getNodesForKey(key, replicaCount)
        nodes.forEach { node ->
            getStore(node).set(key, value, ttlMs)
        }
    }

    fun get(key: String): Any? {
        val nodes = ring.getNodesForKey(key, replicaCount)
        for (node in nodes) {
            try {
                val value = getStore(node).get(key)
                if (value != null) return value
            } catch (e: Exception) {
                // 노드 장애, 다음 노드 시도
                continue
            }
        }
        return null
    }

    private fun getStore(node: ConsistentHashRing.Node): MiniRedis {
        return nodeStores.computeIfAbsent(node.id) { MiniRedis() }
    }
}