// A服务器：比如玩家杀了一个 Boss，奖励了 100 金币
fun onBossKill(player: Player) {
    val current = database.query("SELECT gold FROM player_data WHERE uuid = ?", player.uniqueId)
    val newGold = current + 100
    database.update("UPDATE player_data SET gold = ? WHERE uuid = ?", newGold, player.uniqueId)
    player.sendMessage("获得 100 金币！当前: $newGold")
}

// B服：玩家购买物品，扣除 50 金币
fun onBuyItem(player: Player) {
    val current = database.query("SELECT gold FROM player_data WHERE uuid = ?", player.uniqueId)
    if (current < 50) {
        player.sendMessage("金币不足！")
        return
    }
    val newGold = current - 50
    database.update("UPDATE player_data SET gold = ? WHERE uuid = ?", newGold, player.uniqueId)
    giveItem(player, "钻石剑")
}

时刻 T0: 玩家金币 = 1000

时刻 T1: A服读取 gold = 1000
时刻 T2: B服读取 gold = 1000
时刻 T3: A服写入当前金币为查到的值+100 gold = 1100 (1000 + 100)
时刻 T4: B服写入当前金币为查到的值-50 gold = 950  (1000 - 50)

最终结果: gold = 950

fun transferPlayer(player: Player, targetServer: String) {
    saveToDatabase(player)  // 保存当前数据
    player.connect(targetServer)
}

fun onPlayerJoin(player: Player) {
    val data = loadFromDatabase(player)
    applyToPlayer(player, data)
}

fun updateGold(playerId: UUID, delta: Int) {
    // 先更新缓存
    redis.incrBy("gold:$playerId", delta)
    // 再更新数据库
    database.update("UPDATE player_data SET gold = gold + ? WHERE uuid = ?", delta, playerId)
}

[10:23:45] 玩家 A 金币变更: 1000 -> 1100
[10:23:47] 玩家 A 金币变更: 1100 -> 950

CREATE TABLE player_data (
    uuid VARCHAR(36) PRIMARY KEY,
    gold INT,
    level INT,
    exp BIGINT,
    inventory TEXT,
    skills TEXT,
    quests TEXT,
    ...  -- 几十个字段
);

CREATE TABLE ss_cn1_player_wallet (
    owner_uuid CHAR(36) PRIMARY KEY,
    gold BIGINT NOT NULL,
    diamond BIGINT NOT NULL,
    last_update BIGINT NOT NULL,
    version BIGINT NOT NULL DEFAULT 0,
    fence_token BIGINT NOT NULL DEFAULT 0,
    updated_at BIGINT NOT NULL DEFAULT 0,
    updated_by_server VARCHAR(64) NOT NULL DEFAULT '',
    updated_by_actor VARCHAR(128) NOT NULL DEFAULT ''
);

class DirtyDataContainer {
    private val data = ConcurrentHashMap<String, DataValue<*>>()

    fun <T> get(key: String): T? {
        @Suppress("UNCHECKED_CAST")
        return (data[key] as? DataValue<T>)?.get()
    }

    fun <T> set(key: String, value: T) {
        val wrapper = data.computeIfAbsent(key) {
            if (isDeepType(value)) {
                DataValue.Deep(value)
            } else {
                DataValue.Primitive(value)
            }
        }
        @Suppress("UNCHECKED_CAST")
        (wrapper as DataValue<T>).set(value)
    }

    fun getDirtyKeys(): Set<String> {
        return data.entries
            .filter { it.value.isDirty() }
            .map { it.key }
            .toSet()
    }
}

class Primitive<T>(private var value: T) : DataValue<T>() {
    @Volatile
    private var dirty = false

    override fun set(newValue: T) {
        if (value != newValue) {
            value = newValue
            dirty = true
        }
    }

    override fun isDirty() = dirty
}

class Deep<T>(private val value: T) : DataValue<T>() {
    @Volatile
    private var currentHash = value.hashCode()
    private var lastCleanHash = currentHash

    override fun isDirty(): Boolean {
        val hash = value.hashCode()
        return hash != lastCleanHash
    }

    override fun modify(block: T.() -> T): T {
        val result = value.block()
        currentHash = value.hashCode()
        dirty = true
        return result
    }
}

// 玩家登录，加载数据
val wallet = container.get<Int>("gold") // 1000

// 玩家什么都没做，登出
// 不需要写回数据库，因为没有脏数据

// 修改多个字段
container.set("gold", 1100)
container.set("diamond", 55)
container.set("level", 10)

// 只写入脏字段
val dirtyKeys = container.getDirtyKeys()
// ["gold", "diamond", "level"]

// 批量更新
UPDATE ... SET fields = JSON_SET(fields,
    '$.gold', 1100,
    '$.diamond', 55,
    '$.level', 10
)

interface DataView {
    operator fun <T> get(key: String): T?
    fun <T> getOrDefault(key: String, default: T): T
    fun containsKey(key: String): Boolean
    fun keys(): Set<String>
}

class DataSnapshot(
    private val fields: Map<String, Any?>
) : DataView {
    override fun <T> get(key: String): T? {
        @Suppress("UNCHECKED_CAST")
        return fields[key] as? T
    }

    // 类型安全的 getter
    fun getString(key: String, default: String = ""): String {
        return get<String>(key) ?: default
    }

    fun getInt(key: String, default: Int = 0): Int {
        return get<Number>(key)?.toInt() ?: default
    }
}

fun atomicWrite(
    owner: String,
    domain: String,
    action: (DataEditor) -> Unit
): Boolean {
    val (snapshot, expectedVersion) = readWithVersion(owner, domain)

    val editor = MutableDataEditor(snapshot)
    action(editor)

    val dirtyFields = editor.getDirtyFields()
    if (dirtyFields.isEmpty()) return true

    // CAS 写入
    val sql = """
        UPDATE ss_{region}_{ownerType}_{domain}
        SET <changed_columns>, version = version + 1, updated_at = ?
        WHERE owner_id = ? AND version = ?
    """

    val affected = jdbc.update(
        sql,
        System.currentTimeMillis(),
        ownerId,
        expectedVersion
    )

    return affected > 0  // 返回 true 表示成功
}

fun atomicWriteWithRetry(
    owner: String,
    domain: String,
    maxRetries: Int = 3,
    action: (DataEditor) -> Unit
): TransactionResult {
    repeat(maxRetries) { attempt ->
        val success = atomicWrite(owner, domain, action)
        if (success) {
            return TransactionResult(writes = 1)
        }

        // 冲突，短暂等待后重试
        Thread.sleep(10L * (attempt + 1))
    }

    throw ConcurrentModificationException(
        "CAS failed after $maxRetries retries"
    )
}

public final class APIOperationContext {
    public static APIOperationContext of(String actor, String reason);
    public static APIOperationContext of(String actor, String reason, String traceId);
    public static APIOperationContext of(String actor, String reason, String traceId, String requestId);
}

UUID playerId = player.getUniqueId();
var ns = APIProvider.getAPI()
    .serverSync()
    .data()
    .namespace("economy");

APIOperationContext ctx = APIOperationContext.of(
    "DungeonPlugin",
    "boss_kill_reward",
    "trace:dungeon:run_456",
    "req:dungeon:reward:run_456:" + playerId
);

ns.transaction()
    .player(playerId)
    .context(ctx)
    .write("wallet", editor -> {
        int gold = editor.getOrDefault("gold", 0);
        editor.set("gold", gold + 100);
    })
    .commit();

var ns = APIProvider.getAPI()
    .serverSync()
    .data()
    .namespace("economy");

var wallet = ns.readPlayerDomain(playerId, "wallet");
int gold = wallet.getOrDefault("gold", 0);

Object taxRate = ns.readGlobalValue("market", "global", "tax_rate");

ns.transaction()
    .player(playerId)
    .context(APIOperationContext.of("ShopPlugin", "buy_item"))
    .write("wallet", editor -> {
        int gold = editor.getOrDefault("gold", 0);
        editor.set("gold", gold - 50);
    })
    .commit()
    .handle((result, error) -> {
        if (error != null) {
            System.out.println("提交失败: " + error.getMessage());
            return;
        }
        System.out.println("写入条目: " + result.getWrites());
    });

服务器 A:
  读取 gold = 1000
  计算 newGold = 1100
  写入 gold = 1100

服务器 B:
  读取 gold = 1000
  计算 newGold = 950
  写入 gold = 950

T1: A 读取 version=10, gold=1000
T2: B 读取 version=10, gold=1000
T3: A CAS 写入成功，version=11, gold=1100
T4: B CAS 失败，重试
T5: B 读取 version=11, gold=1100
T6: B CAS 写入成功，version=12, gold=950

fun updateGold(playerId: UUID, delta: Int) {
    val lock = getLock(playerId, "wallet")
    lock.lock()
    try {
        val current = read(playerId, "wallet", "gold")
        val newGold = current + delta
        write(playerId, "wallet", "gold", newGold)
    } finally {
        lock.unlock()
    }
}

T1: A 获取锁
T2: A 读取 gold=1000
T3: A 写入 gold=1100
T4: A 释放锁
T5: B 获取锁
T6: B 读取 gold=1100
T7: B 写入 gold=1050
T8: B 释放锁

val lock = redisson.getLock("player:$playerId:wallet")

// 方式一：固定租约时间
lock.tryLock(5000, 10000, TimeUnit.MILLISECONDS)
// 等待 5 秒，持有 10 秒后自动释放

// 方式二：Watchdog 自动续租
lock.tryLock(5000, TimeUnit.MILLISECONDS)
// 等待 5 秒，持有期间自动续租

val lock = redisson.getLock("player:$playerId:wallet")
lock.tryLock(5000, 10000, TimeUnit.MILLISECONDS)  // 10 秒租约

try {
    val data = loadFromDatabase(playerId)  // 2 秒
    processComplexLogic(data)              // 9 秒
    saveToDatabase(playerId, data)         // 1 秒
    // 总共 12 秒，超过了 10 秒租约！
} finally {
    lock.unlock()
}

T0:  服务器 A 获取锁，租约 10 秒
T10: 租约过期，锁自动释放
T11: 服务器 B 获取锁
T12: 服务器 A 写入数据（以为自己还持有锁）
T13: 服务器 B 写入数据

T0:  服务器 A 获取锁
T5:  你的服务器什么东西卡住主线程了，JVM GC Full Stop，停顿 15 秒
T10: Watchdog 尝试续租，但服务器卡住了
T15: 租约过期，锁自动释放
T16: 服务器 B 获取锁
T20: 服务器 A 恢复，继续执行（以为自己还持有锁）

T0:  服务器 A 获取锁，token=42
T10: 锁过期释放
T11: 服务器 B 获取锁，token=43
T12: 服务器 A 尝试写入，token=42
     数据库检查：42 < 43，拒绝写入
T13: 服务器 B 写入，token=43，成功

object FenceTokenService {
    fun next(owner: String, domain: String): Long {
        val redisKey = RegionContext.fenceKey(owner, domain)
        val fenceKey = "$owner:$domain"

        val redisVal = RedissonManager.client().getAtomicLong(redisKey).incrementAndGet()
        val floor = cachedFloorOrLoad(fenceKey, domain)
        if (redisVal <= floor) {
            val recovered = floor + 1
            RedissonManager.client().getAtomicLong(redisKey).set(recovered)
            return recovered
        }
        return redisVal
    }

    fun updateFloor(fenceKey: String, newValue: Long): Boolean {
        // 在事务内执行: 仅当 current_value < newValue 时更新成功
        // 成功表示 token 仍有效；失败表示有更高 token 已提交
        ...
    }
}

object DataLockManager {
    fun getLock(owner: String, domain: String): RLock {
        return RedissonManager.client().getLock(
            RegionContext.lockKey(owner, domain)
        )
    }

    fun <T> withLock(
        owner: String,
        domain: String,
        leaseTimeMs: Long = -1,  // -1 表示使用 Watchdog
        waitTimeMs: Long = 5_000,
        action: () -> T
    ): T {
        val lock = getLock(owner, domain)
        val acquired = if (leaseTimeMs > 0) {
            lock.tryLock(waitTimeMs, leaseTimeMs, TimeUnit.MILLISECONDS)
        } else {
            lock.tryLock(waitTimeMs, TimeUnit.MILLISECONDS)
        }

        if (!acquired) {
            throw IllegalStateException(
                "无法获取锁: ${RegionContext.lockKey(owner, domain)}"
            )
        }
        try {
            return action()
        } finally {
            if (lock.isHeldByCurrentThread) {
                lock.unlock()
            }
        }
    }
}

DataLockManager.withLock(owner, domain) {
    val token = FenceTokenService.next(owner, domain)
    // ...读取当前行、执行业务修改...
    // ...CAS 更新（where version=? and fence_token < token）...
    FenceTokenService.updateFloor("$owner:$domain", token)
}

fun atomicWriteWithToken(
    owner: String,
    domain: String,
    token: Long,
    fields: Map<String, Any?>
): Boolean {
    val sql = """
        UPDATE ss_{region}_{ownerType}_{domain}
        SET <changed_columns>, version = version + 1,
            fence_token = ?, updated_at = ?
        WHERE owner_id = ? AND fence_token < ?
    """

    val affected = jdbc.update(
        sql,
        token,  // 新 token
        System.currentTimeMillis(),
        ownerId,
        token   // 只有当前 token < 新 token 时才更新
    )

    return affected > 0
}

// 业务代码
fun rewardPlayer(playerId: UUID, gold: Int) {
    DataLockManager.withLock(
        owner = "player:$playerId",
        domain = "wallet",
        leaseTimeMs = -1  // 使用 Watchdog
    ) {
        val current = read(playerId, "wallet", "gold")
        val newGold = current + gold
        write(playerId, "wallet", "gold", newGold)
    }
}

fun updateGold(playerId: UUID, delta: Int) {
    lock.lock()  // 阻塞等待
    try {
        val gold = read("gold")
        write("gold", gold + delta)
    } finally {
        lock.unlock()
    }
}

fun updateGold(playerId: UUID, delta: Int) {
    do {
        val (gold, version) = readWithVersion("gold")
        val newGold = gold + delta
        val success = casWrite("gold", newGold, version)
    } while (!success)
}

CREATE TABLE ss_{region}_{ownerType}_{domain} (
    -- ...其他字段
    version BIGINT NOT NULL DEFAULT 1,
    fence_token BIGINT NOT NULL DEFAULT 0,
    -- ...
);

UPDATE ss_{region}_{ownerType}_{domain}
SET <changed_columns>, version = version + 1
WHERE ... AND version = ?

data class VersionedSnapshot(
    val snapshot: DataSnapshot,
    val version: Long
)

fun readWithVersion(owner: String, domain: String): VersionedSnapshot {
    val row = jdbc.queryForMap(
        "SELECT <domain_columns>, version FROM ss_{region}_{ownerType}_{domain} WHERE ...",
        owner, domain
    )
    return VersionedSnapshot(
        snapshot = DataSnapshot(extractDomainFields(row)),
        version = row["version"] as Long
    )
}

fun casWrite(
    owner: String,
    domain: String,
    fields: Map<String, Any?>,
    expectedVersion: Long
): Boolean {
    val affected = jdbc.update(
        """
        UPDATE ss_{region}_{ownerType}_{domain}
        SET <changed_columns>, version = version + 1
        WHERE ... AND version = ?
        """,
        owner, domain, expectedVersion
    )
    return affected > 0
}

fun atomicWrite(
    owner: String,
    domain: String,
    action: (DataEditor) -> Unit,
    maxRetries: Int = 3
): TransactionResult {
    repeat(maxRetries) { attempt ->
        // 读取当前版本
        val (snapshot, version) = readWithVersion(owner, domain)

        // 执行业务逻辑
        val editor = MutableDataEditor(snapshot)
        action(editor)

        // CAS 写入
        val success = casWrite(owner, domain, editor.toMap(), version)
        if (success) {
            return TransactionResult(writes = 1)
        }

        // 冲突，等待后重试
        Thread.sleep(10L * (attempt + 1))
    }

    throw ConcurrentModificationException(
        "CAS failed after $maxRetries retries"
    )
}

fun exponentialBackoff(attempt: Int): Long {
    val baseDelay = 10L
    val maxDelay = 1000L
    val delay = baseDelay * (1 shl attempt)  // 2^attempt
    return minOf(delay, maxDelay)
}

// 使用
Thread.sleep(exponentialBackoff(attempt))

T1: 线程 A 读取 version=10
T2: 线程 B 修改 version=11
T3: 线程 C 修改 version=10（回退）
T4: 线程 A CAS 写入，version=10 匹配，成功

UPDATE ss_{region}_{ownerType}_{domain}
SET version = version + 1  -- 永远递增
WHERE ... AND version = ?

WHERE version = ? AND fence_token < ?

class CasMetrics {
    private val attempts = AtomicLong(0)
    private val conflicts = AtomicLong(0)

    fun recordAttempt() = attempts.incrementAndGet()
    fun recordConflict() = conflicts.incrementAndGet()

    fun conflictRate(): Double {
        val total = attempts.get()
        return if (total > 0) conflicts.get().toDouble() / total else 0.0
    }
}

CREATE TABLE ss_{region}_outbox (
    event_id VARCHAR(64) PRIMARY KEY,
    aggregate_type VARCHAR(16) NOT NULL,
    aggregate_id VARCHAR(128) NOT NULL,
    domain VARCHAR(64) NOT NULL,
    event_type VARCHAR(64) NOT NULL,
    payload BLOB NOT NULL,
    status VARCHAR(16) NOT NULL,          -- NEW / DISPATCHING / SENT / DEAD
    retry_count INT NOT NULL DEFAULT 0,
    next_retry_at BIGINT NOT NULL,
    created_at BIGINT NOT NULL
);

transaction(db) {
    val updated = domainRepository.casUpdateInTx(
        ownerId = ownerId,
        fields = changedFields,
        expectedVersion = expectedVersion,
        newFenceToken = fenceToken,
        serverName = serverName,
        actor = actor
    )
    if (!updated) error("CAS/fence conflict")

    OutboxRepository.insertInTx(
        eventId = eventId,
        aggregateType = ownerType,
        aggregateId = ownerId,
        domain = domain,
        eventType = "DATA_UPDATED",
        payload = payloadBytes
    )
}

suspend fun pollAndDispatch() {
    reclaimStuckDispatching()
    val events = OutboxRepository.pollNewEventsAll(batchSize) // core + shards
    for (event in events) {
        try {
            publishToTopic("ss:{region}:outbox:${event.domain}", event.payload)
            OutboxRepository.markSent(event.eventId, event.shardKey)
        } catch (e: Exception) {
            if (event.retryCount >= maxRetries) {
                OutboxRepository.markDead(event.eventId, event.shardKey)
            } else {
                OutboxRepository.markRetry(event.eventId, event.retryCount, retryBackoffBase, event.shardKey)
            }
        }
    }
}

var api = APIProvider.getAPI().serverSync();

var request = TransferRequest.builder(player.getUniqueId())
    .targetGroup("dungeon")
    .selectionStrategy(SelectionStrategy.LEAST_PLAYERS)
    .fallbackPolicy(FallbackPolicy.QUEUE)
    .build();

api.transfer().execute(request).handle((result, error) -> {
        if (error != null) {
            player.sendMessage("切服失败：" + error.getMessage());
            return;
        }
        if (result.getCode() == TransferCode.SUCCESS ||
            result.getCode() == TransferCode.CROSS_REGION_SUCCESS) {
            player.sendMessage("正在进入Dungeon服...");
        } else if (result.getCode() == TransferCode.QUEUED) {
            player.sendMessage("当前排队位置：" + result.getQueuePosition());
        } else {
            player.sendMessage("切服失败：" + result.getCode() + " / " + result.getMessage());
        }
    });

public final class ChatRpcBootstrap {

    public static void register() {
        ServerSyncAPI serverSync = APIProvider.getAPI().serverSync();
        ServerSyncRPCAPI rpc = serverSync.rpc();

        // QUERY：查询在线人数
        rpc.onQuery("chat.online.count", ChatOnlineQuery.class, (ctx, req) ->
            new ChatOnlineResp(org.bukkit.Bukkit.getOnlinePlayers().size())
        );

        // COMMAND：执行跨服静音（有副作用，处理器要幂等）
        rpc.onCommand("chat.mute.player", MuteCommand.class, (ctx, req) -> {
            boolean ok = MuteService.mute(req.playerId(), req.minutes(), req.opId());
            return new MuteResult(ok);
        });

        // EVENT：广播聊天提示（无返回）
        rpc.onEvent("chat.notify", ChatNotifyEvent.class, (ctx, req) ->
            org.bukkit.Bukkit.broadcastMessage("[跨服] " + req.message())
        );
    }
}

ServerSyncAPI serverSync = APIProvider.getAPI().serverSync();

// QUERY 到指定服务器
serverSync.rpc()
    .toServer("lobby-1")
    .query(new ChatOnlineQuery("global"), ChatOnlineResp.class, 1500L)
    .thenAccept(resp -> {
        System.out.println("lobby-1 在线: " + resp.online());
    });

// COMMAND 到玩家所在服务器（自动路由）
serverSync.rpc()
    .toPlayer(playerUuid)
    .command(new MuteCommand(playerUuid, 10, "op:20260308:1001"), MuteResult.class, 2500L)
    .handle((resp, error) -> {
        if (error != null) {
            System.out.println("静音失败: " + error.getMessage());
            return;
        }
        System.out.println("静音结果: " + resp.ok());
    });

// EVENT 广播
serverSync.rpc()
    .broadcast()
    .event(new ChatNotifyEvent("全服维护倒计时 5 分钟"));

// queryAll：向指定 tag 组并发查询
serverSync.rpc()
    .toTag("dungeon")
    .queryAll(new DungeonLoadQuery(), DungeonLoadResp.class, 1200L)
    .thenAccept(multi -> {
        multi.getSuccesses().forEach((server, load) ->
            System.out.println(server + " -> " + load.players())
        );
        multi.getFailures().forEach((server, ex) ->
            System.out.println(server + " 查询失败: " + ex.getMessage())
        );
    });

CREATE TABLE ss_{region}_audit_log (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    owner_type VARCHAR(16) NOT NULL,    -- player / global
    owner_id VARCHAR(128) NOT NULL,     -- 玩家 UUID 或聚合 ID
    domain VARCHAR(64) NOT NULL,        -- 如 wallet / profile
    field_key VARCHAR(64) NOT NULL,     -- 如 gold
    old_value TEXT NULL,
    new_value TEXT NULL,
    actor VARCHAR(128) NOT NULL,
    reason VARCHAR(128) NULL,
    trace_id VARCHAR(64) NOT NULL,
    request_id VARCHAR(64) NOT NULL,
    server_name VARCHAR(64) NOT NULL,
    ts BIGINT NOT NULL
);

UUID playerId = player.getUniqueId();
ServerSyncDataNamespaceAPI economy =
    APIProvider.getAPI().serverSync().data().namespace("economy");

APIOperationContext ctx = APIOperationContext.of(
    "DungeonPlugin",
    "boss_kill_reward",
    "trace:dungeon:run_456"
);

economy.transaction()
    .player(playerId)
    .context(ctx)
    .write("wallet", editor -> {
        int oldGold = editor.getOrDefault("gold", 0);
        editor.set("gold", oldGold + 150);
    })
    .commit()
    .handle((result, error) -> {
        if (error != null) {
            System.out.println("奖励写入失败: " + error.getMessage());
            return;
        }
        System.out.println("写入成功，提交条数: " + result.getWrites());
    });

var audit = APIProvider.getAPI().serverSync().audit();

// 1) 按 owner 查最近记录
var ownerLogs = audit.queryByOwner("player", playerId.toString(), "wallet", 50);

// 2) 按 traceId 查整条链路
var traceLogs = audit.queryByTraceId("trace:dungeon:run_456");

// 3) 按时间范围查某域
long now = System.currentTimeMillis();
var rangeLogs = audit.queryByTimeRange(
    "player",
    playerId.toString(),
    "wallet",
    now - 3600_000L,
    now,
    200
);

class WalletReadCache(
    private val ns: ServerSyncDataNamespaceAPI
) {

    private val cache = Caffeine.newBuilder()
        .maximumSize(1000)
        .expireAfterWrite(5, TimeUnit.MINUTES)
        .build<String, DataSnapshot?>()

    fun readPlayerWallet(playerId: UUID): DataSnapshot? {
        val key = "player:$playerId:wallet"
        return cache.get(key) {
            ns.readPlayerDomain(playerId, "wallet")
        }
    }

    fun invalidatePlayerWallet(playerId: UUID) {
        cache.invalidate("player:$playerId:wallet")
    }
}

// 写入后发布事件
redis.getTopic("cache_invalidate").publish(
    CacheInvalidateEvent(owner, domain)
)

// 订阅事件
redis.getTopic("cache_invalidate").addListener { event ->
    cache.invalidate("${event.owner}:${event.domain}")
}

fun batchRead(requests: List<Pair<String, String>>): Map<Pair<String, String>, DataSnapshot> {
    val sql = """
        SELECT owner_id, domain, fields, version
        FROM ss_{region}_{ownerType}_{domain}
        WHERE (owner_id, domain) IN (${requests.joinToString { "(?, ?)" }})
    """

    val results = jdbc.query(sql, ...).associateBy { it.owner to it.domain }
    return results
}

fun batchWrite(updates: List<UpdateRequest>) {
    jdbc.batchUpdate(
        "UPDATE ss_{region}_{ownerType}_{domain} SET <changed_columns>, version = version + 1 WHERE ...",
        updates.map { arrayOf(it.owner, it.domain) }
    )
}