Build a Cross-Server Data Sync Engine
Build a pluggable Kotlin sync engine that keeps any data type consistent across Minecraft servers.
Introduction
30 Second Summary
When thousands of players are online across multiple game servers, every coin earned and item picked up needs to stay consistent. Without smart caching, a busy Minecraft network would drown its database under tens of thousands of writes per second.
In this project, you will build a Kotlin Paper plugin that acts as a cross-server data synchronization engine. Using an interface-based design, the engine can sync any data type (player profiles, guild banks, island progress) with zero database spam, real-time consistency across servers, and safe concurrent access for shared resources.
What You'll Build
A pluggable sync framework built on the SyncableData interface, with a write-back caching layer, MongoDB Change Streams for cross-server cache invalidation, and a session ownership system for conflict-free data access. Any data type that implements the interface gets all of this for free.
By the end of this project, you'll have:
- A SyncableData interface that any data type can implement to plug into the sync engine, with declarative field categorization for deltas, session-owned, and versioned data.
- A write-back cache with delta accumulators that reduces database load by over 99%, batching thousands of in-memory increments into compact bulk operations every 5 seconds.
- A Change Stream listener that invalidates and updates cached data across servers in real time, using hash-based echo filtering to ignore your own writes.
- Secret Mission: Add optimistic retry logic for safe inventory transfers under concurrent modification.
Are there any prerequisites?
This project assumes you are proficient in Kotlin, have basic MongoDB experience, and have built Paper plugins before. You should be comfortable with coroutines, document databases, and the Paper event lifecycle.
Set Up the Project
A Minecraft network with 1000+ concurrent players generates a storm of data writes. Every coin earned, every block broken, every death triggers a database operation.
In this step, you will create the project skeleton and wire up the dependencies that make a write-back cache possible. You will also configure MongoDB as a replica set so that Change Streams can power real-time cross-server communication.
In this step, get ready to:
- Create a Gradle Kotlin DSL project with Paper, MongoDB, and cache dependencies.
- Configure MongoDB as a single-node replica set and verify Change Streams work.
- Create the plugin main class and verify database connectivity.
Create the Gradle Kotlin DSL project
The sync engine needs four core libraries: the Paper API for hooking into Minecraft, the MongoDB Kotlin Coroutine Driver for database access, Caffeine for in-memory caching, and kotlinx-coroutines for async scheduling.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type IntelliJ IDEA and press Enter to open it.
- On the Welcome screen, click New Project. If you already have a project open, go to File then New Project.
- Select Kotlin as the language.
- Select Gradle as the Build system.
- Select Kotlin as the Gradle DSL.
- Name the project sync-engine.
- Expand Advanced Settings and set the GroupId to com.nextwork.syncengine.
- Click Create.
IntelliJ IDEA generates a default build.gradle.kts file. You will replace its contents with the correct dependencies.
- Open build.gradle.kts from the Project tool window on the left (Alt+1).
- Replace the entire file contents with this configuration:
plugins {
kotlin("jvm") version "1.9.0"
}
group = "com.nextwork.syncengine"
version = "1.0-SNAPSHOT"
repositories {
mavenCentral()
maven("https://repo.papermc.io/repository/maven-public/")
}
dependencies {
compileOnly("io.papermc.paper:paper-api:1.21.4-R0.1-SNAPSHOT")
implementation(platform("org.mongodb:mongodb-driver-bom:5.8.0"))
implementation("org.mongodb:mongodb-driver-kotlin-coroutine")
implementation(kotlin("reflect"))
implementation("com.github.ben-manes.caffeine:caffeine:3.2.4")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
}
kotlin {
jvmToolchain(21)
}What does this build file do?
The repositories block adds the Paper Maven repository alongside Maven Central. This lets Gradle find the Paper API artifact.
In dependencies, the Paper API is compileOnly because the server provides it at runtime. The MongoDB BOM manages driver version consistency so you only declare the version once. Caffeine provides the in-memory cache, and kotlinx-coroutines powers the async flush scheduler.
- Save the file.
- Click the Sync Gradle Changes icon that appears in the editor notification bar (or press Ctrl+Shift+O).
- Wait for the Gradle sync to finish. You should see BUILD SUCCESSFUL in the Build tool window at the bottom.
Gradle sync failed?
- Check that your JDK is version 21 or higher. Go to File then Project Structure then Project SDK to verify.
- Make sure you have internet access. Gradle needs to download the dependencies from Maven Central and the Paper repository.
Help me debug my Gradle sync failure
Configure MongoDB as a replica set and create the plugin descriptor
Change Streams require MongoDB to run as a replica set. A single-node replica set is fine for development. You just need to start mongod with the --replSet flag and run rs.initiate() once.
Why does MongoDB need a replica set?
Change Streams rely on the oplog, which is only available in a replica set. A standalone mongod instance does not maintain an oplog, so collection.watch() calls would fail.
A single-node replica set gives you Change Stream support without the overhead of running multiple database nodes. It is the standard approach for local development.
- If MongoDB is currently running as a standalone, stop it first.
- Restart mongod with the replica set flag by running this command in a terminal:
mongod --replSet "rs0" --bind_ip localhost --dbpath /data/dbKeep this terminal open
The mongod process needs to stay running. Open a new terminal tab or window for the next commands.
- Open a second terminal and connect to MongoDB by running:
mongosh- Initiate the replica set by running this command inside mongosh:
rs.initiate()- Verify the replica set is healthy by running:
rs.status()You should see output showing your node with stateStr: "PRIMARY". This confirms the replica set is active.
- Test that Change Streams work by running:
db.test.watch()If no error is thrown and you see a cursor open, Change Streams are working. Press Ctrl+C to close the cursor.
Seeing an error about Change Streams?
- Make sure mongod was started with --replSet "rs0". Without this flag, the oplog is not available.
- Confirm rs.initiate() returned { ok: 1 }. If it returned an error, your mongod may not have the --replSet flag set.
Help me fix my MongoDB replica set configuration
Now create the plugin descriptor that tells the Paper server about your plugin.
- Back in IntelliJ IDEA, right-click the src/main/resources folder in the Project tool window.
- Select New then File.
- Name the file plugin.yml and press Enter.
- Paste the following content into the file:
name: SyncEngine
version: 1.0-SNAPSHOT
main: com.nextwork.syncengine.SyncEnginePlugin
api-version: '1.21'
libraries:
- org.jetbrains.kotlin:kotlin-reflect:1.9.0
- org.mongodb:mongodb-driver-kotlin-coroutine:5.8.0
- com.github.ben-manes.caffeine:caffeine:3.2.4
- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3What does the libraries field do?
Paper's libraries field tells the server to download these dependencies from Maven Central at runtime. This avoids needing to shade (bundle) the jars into your plugin, keeping the plugin file small.
- Save plugin.yml.
Create the main plugin class and verify connectivity
The plugin main class is the entry point. It initializes the MongoClient on server startup and cleanly closes it on shutdown.
- In the Project tool window, right-click src/main/kotlin.
- Select New then Package.
- Enter com.nextwork.syncengine and press Enter.
- Right-click the new syncengine package folder.
- Select New then Kotlin Class/File.
- Choose Class and name it SyncEnginePlugin.
- Replace the generated file contents with the following code:
package com.nextwork.syncengine
import com.mongodb.kotlin.client.coroutine.MongoClient
import org.bukkit.plugin.java.JavaPlugin
class SyncEnginePlugin : JavaPlugin() {
lateinit var mongoClient: MongoClient
private set
override fun onEnable() {
// Create a MongoClient connected to the local replica set
mongoClient = MongoClient.create("mongodb://localhost:27017")
val db = mongoClient.getDatabase("syncengine")
logger.info("SyncEngine enabled. MongoDB connected.")
}
override fun onDisable() {
// Clean up the MongoClient connection on shutdown
mongoClient.close()
logger.info("SyncEngine disabled.")
}
}What does this class do?
- SyncEnginePlugin extends JavaPlugin, which is Paper's base class for all plugins.
- onEnable() runs when the server starts. It creates a MongoClient instance pointing at your local replica set and grabs a reference to the syncengine database.
- onDisable() runs on server shutdown and closes the database connection to prevent resource leaks.
- Save SyncEnginePlugin.kt.
- Verify the project compiles by opening the Gradle tool window (View then Tool Windows then Gradle), expanding Tasks then build, and double-clicking build.
- You should see BUILD SUCCESSFUL in the Run window at the bottom. This confirms all dependencies resolved and the code compiles.
Build failed with unresolved references?
- Make sure plugin.yml is inside src/main/resources, not src/main/kotlin.
- Check that the repositories block in build.gradle.kts includes both mavenCentral() and the Paper Maven URL.
- Confirm your import statement uses com.mongodb.kotlin.client.coroutine.MongoClient, not the synchronous driver import.
Help me fix my build errors in the SyncEngine plugin
✔️ Awesome, I've got everything!
Great. Double check you have saved all three files before moving on.
ⓧ I'd like to double check the full code
Here is the complete code for each file at this point:
plugins {
kotlin("jvm") version "1.9.0"
}
group = "com.nextwork.syncengine"
version = "1.0-SNAPSHOT"
repositories {
mavenCentral()
maven("https://repo.papermc.io/repository/maven-public/")
}
dependencies {
compileOnly("io.papermc.paper:paper-api:1.21.4-R0.1-SNAPSHOT")
implementation(platform("org.mongodb:mongodb-driver-bom:5.8.0"))
implementation("org.mongodb:mongodb-driver-kotlin-coroutine")
implementation(kotlin("reflect"))
implementation("com.github.ben-manes.caffeine:caffeine:3.2.4")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
}
kotlin {
jvmToolchain(21)
}name: SyncEngine
version: 1.0-SNAPSHOT
main: com.nextwork.syncengine.SyncEnginePlugin
api-version: '1.21'
libraries:
- org.jetbrains.kotlin:kotlin-reflect:1.9.0
- org.mongodb:mongodb-driver-kotlin-coroutine:5.8.0
- com.github.ben-manes.caffeine:caffeine:3.2.4
- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3package com.nextwork.syncengine
import com.mongodb.kotlin.client.coroutine.MongoClient
import org.bukkit.plugin.java.JavaPlugin
class SyncEnginePlugin : JavaPlugin() {
lateinit var mongoClient: MongoClient
private set
override fun onEnable() {
mongoClient = MongoClient.create("mongodb://localhost:27017")
val db = mongoClient.getDatabase("syncengine")
logger.info("SyncEngine enabled. MongoDB connected.")
}
override fun onDisable() {
mongoClient.close()
logger.info("SyncEngine disabled.")
}
}Your project skeleton is ready. You have a compiling plugin with a live MongoDB connection and Change Streams enabled. Next up, you will define the data model and build a session ownership system that prevents write conflicts across servers.
Build the Data Model and Session Ownership
Define the SyncableData Interface and Session Ownership
Your plugin connects to MongoDB and the Gradle build compiles cleanly. Now you need to answer a design question: how do you build a sync engine that works for any data type, not just player profiles?
In this step, you will define a SyncableData interface that any data type (player profiles, guild banks, island progress) can implement. Then you will build a concrete SharedProfile as the first implementation, and wire up a session ownership system that prevents two servers from writing to the same document simultaneously.
In this step, get ready to:
- Define the SyncableData interface with @Add, @Put, and @Ignore annotations for declarative field categorization.
- Implement SharedProfile as the first concrete data type with every property annotated.
- Build a SessionManager that claims and releases document ownership using MongoDB atomic operations and a heartbeat.
Define the SyncableData interface
Not all synced data behaves the same way when multiple servers need access. Coins earned on one server can simply be added to the total. Inventory can only be modified by whichever server currently owns the session. Guild settings use optimistic locking.
By defining an interface that declares which fields belong to which category, the entire cache and flush layer can work generically. Any data type that implements this interface automatically gets cross-server sync.
- Create a new package folder for the data model. In your IntelliJ project, right-click the com.nextwork.syncengine package in src/main/kotlin, select New, then Package.
- Name the package model and press Enter.
- Right-click the new model package, select New, then Kotlin Class/File.
- Name the file SyncableData and select File as the type.
- Replace the file contents with the following interface definition:
package com.nextwork.syncengine.model
import kotlin.reflect.KClass
import kotlin.reflect.KParameter
import kotlin.reflect.KProperty1
import kotlin.reflect.full.memberFunctions
import kotlin.reflect.full.memberProperties
/**
* Marks a field as additive (delta accumulator).
* The sync engine uses MongoDB's $inc on these fields.
* Safe for concurrent writes from multiple servers.
*/
@Target(AnnotationTarget.PROPERTY)
@Retention(AnnotationRetention.RUNTIME)
annotation class Add
/**
* Marks a field as a full-value replacement.
* Any server holding the active session can put a new value.
*/
@Target(AnnotationTarget.PROPERTY)
@Retention(AnnotationRetention.RUNTIME)
annotation class Put
/**
* Marks a field as ignored by the sync engine.
* Use for metadata, identifiers, or fields managed outside sync.
*/
@Target(AnnotationTarget.PROPERTY)
@Retention(AnnotationRetention.RUNTIME)
annotation class Ignore
/**
* Any data type that implements this interface gets
* automatic cross-server sync, write-back caching, and
* session ownership protection.
*
* Every property in an implementation MUST be annotated with
* @Add, @Put, or @Ignore. Missing annotations throw at runtime.
*
* All sync methods use Kotlin reflection, so implementations
* only need to declare annotated properties - no manual overrides.
*/
interface SyncableData<ID> {
val id: ID
val version: Long
val contentHash: Int
// Discovers @Add annotations recursively, building dot-notation paths
fun deltaFields(): Set<String> =
discoverFields(this::class, Add::class, "")
// Discovers @Put annotations on top-level properties
fun sessionOwnedFields(): Set<String> =
this::class.memberProperties
.filter { prop -> prop.annotations.any { it is Put } }
.map { it.name }
.toSet()
// Walks a dot-notation path using reflection to read the nested Long value
fun getDeltaValue(field: String): Long {
val parts = field.split(".")
var current: Any? = this
for (part in parts) {
@Suppress("UNCHECKED_CAST")
val prop = current!!::class.memberProperties
.first { it.name == part } as KProperty1<Any, *>
current = prop.get(current)
}
return current as Long
}
// Applies all deltas reflectively by grouping on top-level property
fun withDeltas(deltas: Map<String, Long>): SyncableData<ID> {
if (deltas.isEmpty()) return this
val grouped = deltas.entries.groupBy { it.key.split(".")[0] }
var result: Any = this
for ((topProp, entries) in grouped) {
@Suppress("UNCHECKED_CAST")
val prop = result::class.memberProperties
.first { it.name == topProp } as KProperty1<Any, *>
val currentValue = prop.get(result)
val newValue = if (entries.any { "." in it.key }) {
// Nested data class - recursively apply deltas to sub-fields
applyNestedDeltas(currentValue!!, entries.associate {
it.key.substringAfter(".") to it.value
})
} else {
// Top-level Long field - add the delta directly
(currentValue as Long) + entries.first().value
}
result = reflectiveCopy(result, topProp, newValue)
}
@Suppress("UNCHECKED_CAST")
return result as SyncableData<ID>
}
// Replaces a session-owned field using reflective copy()
fun withSessionOwnedField(field: String, value: Any): SyncableData<ID> {
@Suppress("UNCHECKED_CAST")
return reflectiveCopy(this, field, value) as SyncableData<ID>
}
// Returns a copy with an updated version via reflective copy()
fun withVersion(newVersion: Long): SyncableData<ID> {
@Suppress("UNCHECKED_CAST")
return reflectiveCopy(this, "version", newVersion) as SyncableData<ID>
}
}
// --- Type-safe path operators ---
// Use these instead of raw strings when referencing fields at call sites.
// Top-level: SharedProfile::playtime
// Nested: SharedProfile::currency / Currency::purse
// Builds a dot-notation path from two property references
operator fun <T, R, V> KProperty1<T, R>.div(other: KProperty1<R, V>): String =
"${this.name}.${other.name}"
// Enables chaining for 3+ levels of nesting: A::b / B::c / C::d
operator fun <T, R> String.div(other: KProperty1<T, R>): String =
"$this.${other.name}"
// --- Private helpers ---
// Recursively scans a class for properties with the given annotation
// Throws IllegalStateException if any property is missing @Add, @Put, or @Ignore
private fun discoverFields(
klass: KClass<*>,
annotation: KClass<out Annotation>,
prefix: String
): Set<String> {
val results = mutableSetOf<String>()
for (prop in klass.memberProperties) {
val hasAdd = prop.annotations.any { it is Add }
val hasPut = prop.annotations.any { it is Put }
val hasIgnore = prop.annotations.any { it is Ignore }
if (!hasAdd && !hasPut && !hasIgnore) {
throw IllegalStateException(
"Property '${prop.name}' in ${klass.simpleName} is missing " +
"@Add, @Put, or @Ignore annotation. Every syncable property must declare its strategy."
)
}
val hasTarget = prop.annotations.any { annotation.isInstance(it) }
if (!hasTarget) continue
val path = if (prefix.isEmpty()) prop.name else "$prefix.${prop.name}"
val returnType = prop.returnType.classifier as? KClass<*>
if (returnType != null && returnType.isData) {
results.addAll(discoverFields(returnType, annotation, path))
} else {
results.add(path)
}
}
return results
}
// Calls data class copy() reflectively with a single named parameter changed
private fun reflectiveCopy(instance: Any, propertyName: String, newValue: Any?): Any {
val copyFn = instance::class.memberFunctions.first { it.name == "copy" }
val instanceParam = copyFn.parameters.first { it.kind == KParameter.Kind.INSTANCE }
val targetParam = copyFn.parameters.first { it.name == propertyName }
return copyFn.callBy(mapOf(instanceParam to instance, targetParam to newValue))!!
}
// Recursively applies deltas to nested data class properties
private fun applyNestedDeltas(instance: Any, deltas: Map<String, Long>): Any {
val grouped = deltas.entries.groupBy { it.key.split(".")[0] }
var result = instance
for ((propName, entries) in grouped) {
@Suppress("UNCHECKED_CAST")
val prop = result::class.memberProperties
.first { it.name == propName } as KProperty1<Any, *>
val currentValue = prop.get(result)
val newValue = if (entries.any { "." in it.key }) {
applyNestedDeltas(currentValue!!, entries.associate {
it.key.substringAfter(".") to it.value
})
} else {
(currentValue as Long) + entries.first().value
}
result = reflectiveCopy(result, propName, newValue)
}
return result
}How does reflection-based sync work?
Every property in a SyncableData implementation must declare exactly one of the three annotations. If the scanner finds a property without one, it throws an IllegalStateException at runtime. No silent defaults.
- getDeltaValue() uses KProperty1.get() to walk dot-notation paths reflectively. It splits the path on ., then calls get() on each segment to drill into nested data classes.
- withDeltas() groups deltas by top-level property. For nested paths (like currency.purse), it recursively reconstructs the nested object with the delta applied. reflectiveCopy() calls the data class copy() function via KFunction.callBy() with only the changed parameter.
- withSessionOwnedField() and withVersion() both delegate to reflectiveCopy(). Since data class copy() has all parameters as optional with defaults, passing only the changed one leaves everything else intact.
- contentHash lets the Change Stream listener detect whether an incoming change event is your own write echoing back, so you can skip unnecessary cache invalidation.
- Save SyncableData.kt.
- Confirm the project compiles by clicking Build in the top menu bar, then Build Project. You should see BUILD SUCCESSFUL in the Build output panel at the bottom.
Build failed with unresolved reference?
- Make sure the file is in the correct package folder: src/main/kotlin/com/nextwork/syncengine/model/.
- Check that the package declaration at the top reads package com.nextwork.syncengine.model.
Help me fix my SyncableData build error.
Implement SharedProfile as the first data type
With the interface defined, you need a concrete implementation. SharedProfile represents player data and declares which of its fields are deltas, which are session-owned, and which use versioned access.
- Right-click the model package, select New, then Kotlin Class/File.
- Name the file SharedProfile and select File as the type.
- Replace the file contents with the following data class that implements SyncableData:
package com.nextwork.syncengine.model
// Nested data class for currency with annotated sub-fields
data class Currency(
@Add val purse: Long = 0L,
@Add val bank: Long = 0L
)
// Implements SyncableData with zero manual overrides.
// Kotlin reflection resolves all paths and applies updates automatically.
data class SharedProfile(
@Ignore override val id: String,
// Additive fields (engine recursively discovers nested @Add paths)
@Add val currency: Currency = Currency(),
@Add val playtime: Long = 0L,
@Add val blocksPlaced: Long = 0L,
@Add val blocksBroken: Long = 0L,
@Add val deaths: Long = 0L,
// Put fields (any server with the session can replace these)
@Put val inventory: String = "",
@Put val position: String = "",
@Put val effects: List<String> = emptyList(),
// Ignored by sync (versioned or managed externally)
@Ignore val stats: Map<String, Long> = emptyMap(),
@Ignore val pets: List<String> = emptyList(),
@Ignore val playerClass: String = "default",
@Ignore override val version: Long = 0L,
@Ignore override val contentHash: Int = 0
) : SyncableData<String>Why reflection instead of manual overrides?
Without reflection, every SyncableData implementation would need to manually code getDeltaValue(), withDeltas(), withSessionOwnedField(), and withVersion() with hardcoded when expressions listing every field path. Adding a new field means updating four methods. That does not scale.
With kotlin-reflect, the interface handles everything generically. A new data type just declares annotated properties and implements SyncableData. The reflection-based default methods discover fields, resolve paths, and apply updates automatically. Zero boilerplate per implementation.
- Save SharedProfile.kt.
- Build the project (Build > Build Project). You should see BUILD SUCCESSFUL in the output panel.
Build the SessionManager
On a multi-server Minecraft network, multiple servers might need access to the same document at the same time. A player could disconnect from Server A and join Server B within seconds. A guild officer on one server could edit the guild bank while another server is processing a transaction.
The SessionManager solves this with a generic ownership claim system. A server must successfully claim a session for a specific document and data type before it can write session-owned fields. The claim uses MongoDB's findOneAndUpdate with a filter that only succeeds if no other server holds an active session for that document.
Why findOneAndUpdate instead of find-then-update?
A find followed by a separate update has a race condition. Two servers could both read "no active session", then both write their claim. findOneAndUpdate is atomic at the database level. The filter and the update execute as one operation, so only one server can ever win the claim.
- Create a new package called session inside com.nextwork.syncengine (same way you created model).
- Inside the session package, create a new Kotlin file named SessionManager.
- Add the following code to define the class skeleton, TTL index setup, and the claim/release methods:
package com.nextwork.syncengine.session
import com.mongodb.client.model.Filters
import com.mongodb.client.model.FindOneAndUpdateOptions
import com.mongodb.client.model.IndexOptions
import com.mongodb.client.model.ReturnDocument
import com.mongodb.client.model.Updates
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import kotlinx.coroutines.*
import org.bson.Document
import java.util.concurrent.TimeUnit
class SessionManager(
private val db: MongoDatabase,
private val serverId: String
) {
private val sessions = db.getCollection<Document>("sessions")
private val heartbeatJobs = mutableMapOf<String, Job>()
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
// Creates a TTL index so MongoDB automatically deletes expired sessions
suspend fun init() {
sessions.createIndex(
Document("expiresAt", 1),
IndexOptions().expireAfter(0, TimeUnit.SECONDS)
)
}
// Builds a composite key so each data type tracks sessions independently
private fun sessionKey(dataType: String, documentId: String): String =
"$dataType:$documentId"
// Atomically claims a session for any SyncableData document
suspend fun claimSession(dataType: String, documentId: String): Boolean {
val key = sessionKey(dataType, documentId)
val now = System.currentTimeMillis()
val expiry = now + 60_000L
val result = sessions.findOneAndUpdate(
Filters.and(
Filters.eq("_id", key),
Filters.or(
Filters.exists("_id", false),
Filters.lt("expiresAt", now)
)
),
Updates.combine(
Updates.set("serverId", serverId),
Updates.set("dataType", dataType),
Updates.set("documentId", documentId),
Updates.set("claimedAt", now),
Updates.set("expiresAt", expiry)
),
FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER)
) ?: return false
startHeartbeat(key)
return result.getString("serverId") == serverId
}
// Releases the session and stops the heartbeat for this document
suspend fun releaseSession(dataType: String, documentId: String) {
val key = sessionKey(dataType, documentId)
heartbeatJobs[key]?.cancel()
heartbeatJobs.remove(key)
sessions.deleteOne(Filters.and(
Filters.eq("_id", key),
Filters.eq("serverId", serverId)
))
}
// Placeholder - heartbeat implementation added next
private fun startHeartbeat(key: String) {
}
fun shutdown() {
scope.cancel()
}
}What does this code do?
- sessionKey() creates a composite key from the data type and document ID (e.g. "profile:player123" or "guildbank:guild456"). This means a server can hold a session on a player's profile and a guild bank simultaneously without conflicts.
- init() creates a TTL index on the expiresAt field. MongoDB's TTL mechanism automatically deletes documents once their timestamp passes, so expired sessions disappear without any cleanup code.
- claimSession() takes a dataType and documentId, builds the composite key, and uses findOneAndUpdate with a filter requiring either no existing document or an expired one. With upsert(true), if no document matches, MongoDB creates one. The return check confirms this server won the claim.
- releaseSession() cancels the heartbeat coroutine and deletes the session document. The filter ensures you only delete your own session, never another server's.
- shutdown() cancels all coroutines in the scope. You will call this from the plugin's onDisable() later.
- Save SessionManager.kt.
- Build the project (Build > Build Project). You should see BUILD SUCCESSFUL in the output panel.
Seeing unresolved reference errors for MongoDB classes?
- Make sure your build.gradle.kts includes the MongoDB BOM and coroutine driver dependencies from Step 1. Click the Gradle sync icon (elephant with a refresh arrow) in the toolbar to reload.
- Verify the imports at the top of the file match exactly. The Filters, Updates, and FindOneAndUpdateOptions all come from com.mongodb.client.model.
Help me fix my SessionManager MongoDB import errors.
Implement the session heartbeat
The claim system has a gap. If a server crashes, it never calls releaseSession(). The session document stays in MongoDB forever, and no other server can claim the profile.
The fix is a heartbeat. Every 30 seconds, the owning server pushes the expiresAt timestamp forward by 60 seconds. If the server crashes, the heartbeat stops, the TTL expires, and MongoDB automatically deletes the stale session.
- In SessionManager.kt, find the empty startHeartbeat function:
// Placeholder - heartbeat implementation added next
private fun startHeartbeat(key: String) {
}- Replace the entire startHeartbeat function with this implementation:
// Extends the session TTL every 30 seconds to prove this server is alive
private fun startHeartbeat(key: String) {
heartbeatJobs[key] = scope.launch {
while (isActive) {
delay(30_000L)
val expiry = System.currentTimeMillis() + 60_000L
sessions.updateOne(
Filters.and(
Filters.eq("_id", key),
Filters.eq("serverId", serverId)
),
Updates.set("expiresAt", expiry)
)
}
}
}What does the heartbeat do?
- scope.launch starts a background coroutine that runs independently of the main thread.
- Every 30 seconds, it pushes expiresAt 60 seconds into the future. The 30-second interval with a 60-second TTL gives one missed heartbeat as a safety margin before the session expires.
- The filter includes serverId to ensure you only extend your own session. If another server somehow claimed the session, this update silently does nothing.
- The key parameter is the composite "dataType:documentId" string. This means a server running heartbeats for a profile and a guild bank tracks them as separate entries, with no collisions.
- while (isActive) stops the loop when the coroutine is cancelled (during releaseSession or shutdown).
- Save SessionManager.kt.
- Build the project again (Build > Build Project). You should see BUILD SUCCESSFUL with no errors.
Build error about suspend function in non-suspend context?
- The sessions.updateOne() call inside the coroutine is valid because scope.launch provides a coroutine context. Make sure updateOne is inside the scope.launch { } block, not outside it.
- Check that your kotlinx-coroutines-core dependency is present in build.gradle.kts and click the Gradle sync icon.
Help me fix my coroutine build error in SessionManager.
✔️ Awesome, I've got everything!
Great. Double check you have saved SyncableData.kt, SharedProfile.kt, and SessionManager.kt.
ⓧ I'd like to double check the full code
Here is the complete SyncableData.kt:
package com.nextwork.syncengine.model
import kotlin.reflect.KClass
import kotlin.reflect.KParameter
import kotlin.reflect.KProperty1
import kotlin.reflect.full.memberFunctions
import kotlin.reflect.full.memberProperties
@Target(AnnotationTarget.PROPERTY)
@Retention(AnnotationRetention.RUNTIME)
annotation class Add
@Target(AnnotationTarget.PROPERTY)
@Retention(AnnotationRetention.RUNTIME)
annotation class Put
@Target(AnnotationTarget.PROPERTY)
@Retention(AnnotationRetention.RUNTIME)
annotation class Ignore
interface SyncableData<ID> {
val id: ID
val version: Long
val contentHash: Int
fun deltaFields(): Set<String> =
discoverFields(this::class, Add::class, "")
fun sessionOwnedFields(): Set<String> =
this::class.memberProperties
.filter { prop -> prop.annotations.any { it is Put } }
.map { it.name }
.toSet()
fun getDeltaValue(field: String): Long {
val parts = field.split(".")
var current: Any? = this
for (part in parts) {
@Suppress("UNCHECKED_CAST")
val prop = current!!::class.memberProperties
.first { it.name == part } as KProperty1<Any, *>
current = prop.get(current)
}
return current as Long
}
fun withDeltas(deltas: Map<String, Long>): SyncableData<ID> {
if (deltas.isEmpty()) return this
val grouped = deltas.entries.groupBy { it.key.split(".")[0] }
var result: Any = this
for ((topProp, entries) in grouped) {
@Suppress("UNCHECKED_CAST")
val prop = result::class.memberProperties
.first { it.name == topProp } as KProperty1<Any, *>
val currentValue = prop.get(result)
val newValue = if (entries.any { "." in it.key }) {
applyNestedDeltas(currentValue!!, entries.associate {
it.key.substringAfter(".") to it.value
})
} else {
(currentValue as Long) + entries.first().value
}
result = reflectiveCopy(result, topProp, newValue)
}
@Suppress("UNCHECKED_CAST")
return result as SyncableData<ID>
}
fun withSessionOwnedField(field: String, value: Any): SyncableData<ID> {
@Suppress("UNCHECKED_CAST")
return reflectiveCopy(this, field, value) as SyncableData<ID>
}
fun withVersion(newVersion: Long): SyncableData<ID> {
@Suppress("UNCHECKED_CAST")
return reflectiveCopy(this, "version", newVersion) as SyncableData<ID>
}
}
operator fun <T, R, V> KProperty1<T, R>.div(other: KProperty1<R, V>): String =
"${this.name}.${other.name}"
operator fun <T, R> String.div(other: KProperty1<T, R>): String =
"$this.${other.name}"
private fun discoverFields(
klass: KClass<*>,
annotation: KClass<out Annotation>,
prefix: String
): Set<String> {
val results = mutableSetOf<String>()
for (prop in klass.memberProperties) {
val hasAdd = prop.annotations.any { it is Add }
val hasPut = prop.annotations.any { it is Put }
val hasIgnore = prop.annotations.any { it is Ignore }
if (!hasAdd && !hasPut && !hasIgnore) {
throw IllegalStateException(
"Property '${prop.name}' in ${klass.simpleName} is missing " +
"@Add, @Put, or @Ignore annotation."
)
}
val hasTarget = prop.annotations.any { annotation.isInstance(it) }
if (!hasTarget) continue
val path = if (prefix.isEmpty()) prop.name else "$prefix.${prop.name}"
val returnType = prop.returnType.classifier as? KClass<*>
if (returnType != null && returnType.isData) {
results.addAll(discoverFields(returnType, annotation, path))
} else {
results.add(path)
}
}
return results
}
private fun reflectiveCopy(instance: Any, propertyName: String, newValue: Any?): Any {
val copyFn = instance::class.memberFunctions.first { it.name == "copy" }
val instanceParam = copyFn.parameters.first { it.kind == KParameter.Kind.INSTANCE }
val targetParam = copyFn.parameters.first { it.name == propertyName }
return copyFn.callBy(mapOf(instanceParam to instance, targetParam to newValue))!!
}
private fun applyNestedDeltas(instance: Any, deltas: Map<String, Long>): Any {
val grouped = deltas.entries.groupBy { it.key.split(".")[0] }
var result = instance
for ((propName, entries) in grouped) {
@Suppress("UNCHECKED_CAST")
val prop = result::class.memberProperties
.first { it.name == propName } as KProperty1<Any, *>
val currentValue = prop.get(result)
val newValue = if (entries.any { "." in it.key }) {
applyNestedDeltas(currentValue!!, entries.associate {
it.key.substringAfter(".") to it.value
})
} else {
(currentValue as Long) + entries.first().value
}
result = reflectiveCopy(result, propName, newValue)
}
return result
}Here is the complete SharedProfile.kt:
package com.nextwork.syncengine.model
data class Currency(
@Add val purse: Long = 0L,
@Add val bank: Long = 0L
)
data class SharedProfile(
@Ignore override val id: String,
@Add val currency: Currency = Currency(),
@Add val playtime: Long = 0L,
@Add val blocksPlaced: Long = 0L,
@Add val blocksBroken: Long = 0L,
@Add val deaths: Long = 0L,
@Put val inventory: String = "",
@Put val position: String = "",
@Put val effects: List<String> = emptyList(),
@Ignore val stats: Map<String, Long> = emptyMap(),
@Ignore val pets: List<String> = emptyList(),
@Ignore val playerClass: String = "default",
@Ignore override val version: Long = 0L,
@Ignore override val contentHash: Int = 0
) : SyncableData<String>Here is the complete SessionManager.kt:
package com.nextwork.syncengine.session
import com.mongodb.client.model.Filters
import com.mongodb.client.model.FindOneAndUpdateOptions
import com.mongodb.client.model.IndexOptions
import com.mongodb.client.model.ReturnDocument
import com.mongodb.client.model.Updates
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import kotlinx.coroutines.*
import org.bson.Document
import java.util.concurrent.TimeUnit
class SessionManager(
private val db: MongoDatabase,
private val serverId: String
) {
private val sessions = db.getCollection<Document>("sessions")
private val heartbeatJobs = mutableMapOf<String, Job>()
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
suspend fun init() {
sessions.createIndex(
Document("expiresAt", 1),
IndexOptions().expireAfter(0, TimeUnit.SECONDS)
)
}
private fun sessionKey(dataType: String, documentId: String): String =
"$dataType:$documentId"
suspend fun claimSession(dataType: String, documentId: String): Boolean {
val key = sessionKey(dataType, documentId)
val now = System.currentTimeMillis()
val expiry = now + 60_000L
val result = sessions.findOneAndUpdate(
Filters.and(
Filters.eq("_id", key),
Filters.or(
Filters.exists("_id", false),
Filters.lt("expiresAt", now)
)
),
Updates.combine(
Updates.set("serverId", serverId),
Updates.set("dataType", dataType),
Updates.set("documentId", documentId),
Updates.set("claimedAt", now),
Updates.set("expiresAt", expiry)
),
FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER)
) ?: return false
startHeartbeat(key)
return result.getString("serverId") == serverId
}
suspend fun releaseSession(dataType: String, documentId: String) {
val key = sessionKey(dataType, documentId)
heartbeatJobs[key]?.cancel()
heartbeatJobs.remove(key)
sessions.deleteOne(Filters.and(
Filters.eq("_id", key),
Filters.eq("serverId", serverId)
))
}
private fun startHeartbeat(key: String) {
heartbeatJobs[key] = scope.launch {
while (isActive) {
delay(30_000L)
val expiry = System.currentTimeMillis() + 60_000L
sessions.updateOne(
Filters.and(
Filters.eq("_id", key),
Filters.eq("serverId", serverId)
),
Updates.set("expiresAt", expiry)
)
}
}
}
fun shutdown() {
scope.cancel()
}
}Your interface is defined, a concrete implementation is in place, and sessions are protected. Next up, you will build the write-back cache that works with any SyncableData type, accumulating changes in memory and flushing them to MongoDB in efficient batches.
Build the Write-Back Cache with Delta Accumulators
Your session system ensures only one server owns a profile at a time. But even with ownership settled, writing every tiny change to MongoDB immediately creates a massive bottleneck.
Every coin earned, every block broken, every second of playtime generates a database write. Across 1000 players, that is potentially 20,000 writes per second. A write-back cache solves this by accumulating changes in memory and flushing them to the database in batches.
In this step, get ready to:
- Create a DeltaAccumulator that tracks pending changes as atomic counters.
- Build a ProfileCache that serves profiles with accumulated deltas applied on read.
- Build a FlushScheduler that batches dirty profiles into bulk MongoDB writes every 5 seconds.
Create the DeltaAccumulator class
Before building the cache itself, you need a data structure that can safely track pending increments from multiple threads. The DeltaAccumulator uses Java's AtomicLong fields so the game thread and flush thread can both access counters without locking.
- In the Project panel on the left side of IntelliJ, right-click the syncengine package (under src/main/kotlin/com/nextwork/syncengine).
- Select New, then Package.
- Enter cache as the package name and press Enter.
- Right-click the new cache package, select New then Kotlin Class/File.
- Enter DeltaAccumulator and select Class.
- Replace the generated file contents with the following code:
package com.nextwork.syncengine.cache
import kotlin.reflect.KProperty1
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.ConcurrentHashMap
// A generic accumulator that works with any SyncableData type
class DeltaAccumulator(deltaFields: Set<String>) {
// One AtomicLong per declared delta field
private val counters = ConcurrentHashMap<String, AtomicLong>()
init {
// Pre-populate counters for all declared delta fields
deltaFields.forEach { counters[it] = AtomicLong(0) }
}
// Safely increment using a dot-notation path string
fun increment(field: String, amount: Long) {
counters[field]?.addAndGet(amount)
}
// Type-safe increment using a property reference (top-level fields)
fun increment(prop: KProperty1<*, *>, amount: Long) {
counters[prop.name]?.addAndGet(amount)
}
// Returns true if any counter has a pending delta
fun isDirty(): Boolean =
counters.values.any { it.get() != 0L }
// Peeks at current counter values without draining (for read-time overlay)
fun currentToMap(): Map<String, Long> =
counters.mapValues { it.value.get() }
.filter { it.value != 0L }
// Atomically drains all counters to zero and returns the values
fun drainToMap(): Map<String, Long> =
counters.mapValues { it.value.getAndSet(0) }
.filter { it.value != 0L }
// Re-adds deltas on flush failure so no data is lost
fun restore(deltas: Map<String, Long>) {
deltas.forEach { (field, value) ->
counters[field]?.addAndGet(value)
}
}
}What does this code do?
Instead of hardcoding individual fields, the accumulator takes a Set<String> of delta field names from the SyncableData interface and creates one AtomicLong per field. This makes it work for any data type.
The isDirty() method tells the flush scheduler whether this entry has pending changes worth writing. drainToMap() atomically resets all counters to zero and returns the values that were pending. restore() re-adds deltas if the flush fails, ensuring no data is lost.
- Save DeltaAccumulator.kt with Cmd+S (macOS) or Ctrl+S (Windows).
- Check that IntelliJ shows no red underlines in the file. The file should have no errors since AtomicLong is part of the Java standard library.
Seeing unresolved reference errors?
Make sure the package declaration at the top reads com.nextwork.syncengine.cache and matches the folder path you created.
If IntelliJ still shows red, try File then Invalidate Caches / Restart to force a re-index.
help me fix unresolved reference errors in my DeltaAccumulator class
Build the SyncDataCache with Caffeine
The SyncDataCache is the central piece of the write-back caching layer. It uses Caffeine (a high-performance Java caching library) to store data snapshots in memory with automatic eviction. Because it works with the SyncableData interface, any data type plugs in without changes.
On a cache miss, it loads the document from MongoDB. On reads, it overlays the pending deltas from the accumulator using withDeltas() so callers always see the latest totals without touching the database.
- Right-click the cache package, select New then Kotlin Class/File.
- Enter SyncDataCache and select Class.
- Replace the generated file contents with the following code:
package com.nextwork.syncengine.cache
import com.github.benmanes.caffeine.cache.Caffeine
import com.mongodb.client.model.Filters
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.model.SharedProfile
import com.nextwork.syncengine.model.SyncableData
import kotlin.reflect.KProperty1
import kotlinx.coroutines.runBlocking
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
class ProfileCache(private val db: MongoDatabase) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val accumulators = ConcurrentHashMap<String, DeltaAccumulator>()
// Caffeine cache with auto-load from MongoDB on miss
private val cache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterAccess(10, TimeUnit.MINUTES)
.build<String, SharedProfile> { id ->
runBlocking {
profiles.find(Filters.eq("_id", id)).firstOrNull()
?: SharedProfile(id = id)
}
}
// Get or create the accumulator for a profile (uses deltaFields() from SyncableData)
fun getAccumulator(profileId: String): DeltaAccumulator =
accumulators.getOrPut(profileId) {
DeltaAccumulator(SharedProfile(id = profileId).deltaFields())
}
// Type-safe increment using the / operator for nested paths
// Usage: addDelta(id, SharedProfile::currency / Currency::purse, 100L)
fun addDelta(profileId: String, field: String, amount: Long) {
getAccumulator(profileId).increment(field, amount)
}
// Type-safe increment for top-level properties
// Usage: addDelta(id, SharedProfile::playtime, 5L)
fun addDelta(profileId: String, prop: KProperty1<*, *>, amount: Long) {
getAccumulator(profileId).increment(prop, amount)
}
// Return the cached snapshot with accumulated deltas applied via reflection
fun getProfile(profileId: String): SharedProfile {
val base = cache.get(profileId)!!
val acc = accumulators[profileId] ?: return base
val deltas = acc.currentToMap()
if (deltas.isEmpty()) return base
@Suppress("UNCHECKED_CAST")
return base.withDeltas(deltas) as SharedProfile
}
// Replace a cached entry (used by Change Stream listener)
fun updateCached(profile: SharedProfile) {
cache.put(profile.id, profile)
}
// Return all accumulators that have pending writes
fun getAllDirtyAccumulators(): Map<String, DeltaAccumulator> =
accumulators.filter { it.value.isDirty() }
}What does this code do?
The Caffeine cache holds up to 10,000 profiles in memory and evicts any profile not accessed for 10 minutes. When a profile is requested that is not in the cache, the build lambda loads it from MongoDB automatically.
The getAccumulator() method calls deltaFields() on a fresh SharedProfile instance to discover all @Add fields via reflection. This means the accumulator automatically knows about every delta field without any manual listing.
The addDelta() method has two overloads. You can pass a string path (from the / operator) for nested fields, or a KProperty1 reference for top-level fields. Both are type-safe at the call site:
// Top-level field - property reference
cache.addDelta(playerId, SharedProfile::playtime, 1L)
// Nested field - / operator builds "currency.purse"
cache.addDelta(playerId, SharedProfile::currency / Currency::purse, 100L)The getProfile() method calls currentToMap() on the accumulator (peeks without draining) and passes the result to withDeltas() from the SyncableData interface. The reflection engine applies all pending deltas to the cached snapshot in one call. No when blocks, no manual copy() calls.
- Save ProfileCache.kt.
- Check that IntelliJ resolves all imports with no red underlines. The Caffeine and MongoDB driver dependencies are already in your build.gradle.kts.
Caffeine import not resolving?
If com.github.benmanes.caffeine shows as red, IntelliJ might not have synced your Gradle dependencies yet.
Click the Gradle elephant icon in the top-right notification bar, or open the Gradle tool window and click the refresh icon to re-sync.
help me resolve Caffeine import issues in my Kotlin project
✔️ Awesome, I've got everything!
Great. Make sure both DeltaAccumulator.kt and ProfileCache.kt are saved.
ⓧ I'd like to double check the full code
Here is the full DeltaAccumulator.kt:
package com.nextwork.syncengine.cache
import kotlin.reflect.KProperty1
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.ConcurrentHashMap
class DeltaAccumulator(deltaFields: Set<String>) {
private val counters = ConcurrentHashMap<String, AtomicLong>()
init {
deltaFields.forEach { counters[it] = AtomicLong(0) }
}
fun increment(field: String, amount: Long) {
counters[field]?.addAndGet(amount)
}
fun increment(prop: KProperty1<*, *>, amount: Long) {
counters[prop.name]?.addAndGet(amount)
}
fun isDirty(): Boolean =
counters.values.any { it.get() != 0L }
fun currentToMap(): Map<String, Long> =
counters.mapValues { it.value.get() }
.filter { it.value != 0L }
fun drainToMap(): Map<String, Long> =
counters.mapValues { it.value.getAndSet(0) }
.filter { it.value != 0L }
fun restore(deltas: Map<String, Long>) {
deltas.forEach { (field, value) ->
counters[field]?.addAndGet(value)
}
}
}Here is the full ProfileCache.kt:
package com.nextwork.syncengine.cache
import com.github.benmanes.caffeine.cache.Caffeine
import com.mongodb.client.model.Filters
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.model.SharedProfile
import com.nextwork.syncengine.model.SyncableData
import kotlin.reflect.KProperty1
import kotlinx.coroutines.runBlocking
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
class ProfileCache(private val db: MongoDatabase) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val accumulators = ConcurrentHashMap<String, DeltaAccumulator>()
private val cache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterAccess(10, TimeUnit.MINUTES)
.build<String, SharedProfile> { id ->
runBlocking {
profiles.find(Filters.eq("_id", id)).firstOrNull()
?: SharedProfile(id = id)
}
}
fun getAccumulator(profileId: String): DeltaAccumulator =
accumulators.getOrPut(profileId) {
DeltaAccumulator(SharedProfile(id = profileId).deltaFields())
}
fun addDelta(profileId: String, field: String, amount: Long) {
getAccumulator(profileId).increment(field, amount)
}
fun addDelta(profileId: String, prop: KProperty1<*, *>, amount: Long) {
getAccumulator(profileId).increment(prop, amount)
}
fun getProfile(profileId: String): SharedProfile {
val base = cache.get(profileId)!!
val acc = accumulators[profileId] ?: return base
val deltas = acc.currentToMap()
if (deltas.isEmpty()) return base
@Suppress("UNCHECKED_CAST")
return base.withDeltas(deltas) as SharedProfile
}
fun updateCached(profile: SharedProfile) {
cache.put(profile.id, profile)
}
fun getAllDirtyAccumulators(): Map<String, DeltaAccumulator> =
accumulators.filter { it.value.isDirty() }
}Build the FlushScheduler
The cache accumulates deltas, but those changes live only in memory. If the server crashes before writing to MongoDB, those deltas vanish. The FlushScheduler solves this by running a coroutine that wakes up every 5 seconds, collects all dirty accumulators, and writes them to MongoDB in a single bulkWrite operation.
- Right-click the cache package, select New then Kotlin Class/File.
- Enter FlushScheduler and select Class.
- Replace the generated file contents with the following code:
package com.nextwork.syncengine.cache
import com.mongodb.client.model.BulkWriteOptions
import com.mongodb.client.model.Filters
import com.mongodb.client.model.UpdateOneModel
import com.mongodb.client.model.Updates
import com.mongodb.client.model.WriteModel
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.model.SharedProfile
import kotlinx.coroutines.*
import org.bson.Document
class FlushScheduler(
private val db: MongoDatabase,
private val cache: ProfileCache
) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
// Start the periodic flush loop
fun start() {
scope.launch {
while (isActive) {
delay(5_000L)
flush()
}
}
}
private suspend fun flush() {
val dirty = cache.getAllDirtyAccumulators()
if (dirty.isEmpty()) return
val writes = mutableListOf<WriteModel<SharedProfile>>()
val drainedDeltas = mutableMapOf<String, Map<String, Long>>()
// Build one UpdateOneModel per dirty profile
for ((profileId, acc) in dirty) {
val deltas = acc.drainToMap()
if (deltas.isEmpty()) continue
drainedDeltas[profileId] = deltas
// Use $inc for each accumulated delta field
val incUpdates = deltas.map { (field, value) -> Updates.inc(field, value) }
val update = Updates.combine(incUpdates + Updates.inc("version", 1))
writes.add(
UpdateOneModel(
Filters.eq("_id", profileId),
update
)
)
}
if (writes.isNotEmpty()) {
try {
// Send all writes in a single unordered bulk operation
profiles.bulkWrite(writes, BulkWriteOptions().ordered(false))
} catch (e: Exception) {
// Re-add deltas on failure so they are not lost
for ((profileId, deltas) in drainedDeltas) {
for ((field, value) in deltas) {
cache.addDelta(profileId, field, value)
}
}
}
}
}
fun shutdown() {
scope.cancel()
}
}What does this code do?
Every 5 seconds, flush() collects all accumulators with non-zero deltas. It drains each accumulator (atomically resetting counters to zero) and builds an UpdateOneModel per profile.
Each model uses $inc to atomically add the accumulated values and bumps the version field. All the models are sent as a single unordered bulkWrite to minimize round trips.
If the bulk write fails, the drained deltas are re-added to their accumulators. This ensures no data is lost even during transient network failures.
💡 Why unordered bulk writes?
Setting ordered(false) tells MongoDB it can execute the writes in any order and in parallel. Since each write targets a different profile document, there is no dependency between them. This maximizes throughput for the batch.
- Save FlushScheduler.kt.
- Verify IntelliJ shows no red underlines. All MongoDB driver imports should resolve from the BOM dependency in build.gradle.kts.
WriteModel or BulkWriteOptions not resolving?
These classes come from the MongoDB driver core module, which is included transitively via the BOM. If they still show as red, re-sync Gradle by clicking the elephant icon in the notification bar.
Also confirm your build.gradle.kts has implementation(platform("org.mongodb:mongodb-driver-bom:5.8.0")) before the coroutine driver dependency.
help me fix MongoDB driver import issues in my FlushScheduler
✔️ Awesome, I've got everything!
Great. Make sure FlushScheduler.kt is saved and compiling cleanly.
ⓧ I'd like to double check the full code
Here is the full FlushScheduler.kt:
package com.nextwork.syncengine.cache
import com.mongodb.client.model.BulkWriteOptions
import com.mongodb.client.model.Filters
import com.mongodb.client.model.UpdateOneModel
import com.mongodb.client.model.Updates
import com.mongodb.client.model.WriteModel
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.model.SharedProfile
import kotlinx.coroutines.*
import org.bson.Document
class FlushScheduler(
private val db: MongoDatabase,
private val cache: ProfileCache
) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
fun start() {
scope.launch {
while (isActive) {
delay(5_000L)
flush()
}
}
}
private suspend fun flush() {
val dirty = cache.getAllDirtyAccumulators()
if (dirty.isEmpty()) return
val writes = mutableListOf<WriteModel<SharedProfile>>()
val drainedDeltas = mutableMapOf<String, Map<String, Long>>()
for ((profileId, acc) in dirty) {
val deltas = acc.drainToMap()
if (deltas.isEmpty()) continue
drainedDeltas[profileId] = deltas
val incUpdates = deltas.map { (field, value) -> Updates.inc(field, value) }
val update = Updates.combine(incUpdates + Updates.inc("version", 1))
writes.add(
UpdateOneModel(
Filters.eq("_id", profileId),
update
)
)
}
if (writes.isNotEmpty()) {
try {
profiles.bulkWrite(writes, BulkWriteOptions().ordered(false))
} catch (e: Exception) {
// Re-add deltas on failure
for ((profileId, deltas) in drainedDeltas) {
for ((field, value) in deltas) {
cache.addDelta(profileId, field, value)
}
}
}
}
}
fun shutdown() {
scope.cancel()
}
}Your write-back cache is complete. Player actions now accumulate as fast in-memory deltas, and the FlushScheduler batches them into efficient bulk writes on a 5-second cadence. Next up, you'll wire up MongoDB Change Streams so other servers instantly learn about these flushed updates.
Implement Change Streams for Cross-Server Invalidation
Your write-back cache now batches flushes beautifully on a single server. But what happens when Server B flushes a profile that Server A has cached?
Server A's cache goes stale immediately. Without a notification system, it keeps serving outdated data until the Caffeine TTL expires. MongoDB Change Streams solve this by pushing real-time notifications whenever a document changes. You will open a stream, filter for relevant operations, and use hash-based echo filtering to avoid reacting to your own writes.
In this step, get ready to:
- Open a Change Stream with an aggregation pipeline that filters for insert, update, and replace operations.
- Implement hash-based echo filtering so a server ignores its own flushed writes.
- Add resilience with resume tokens and automatic reconnection on stream failure.
Open a Change Stream on the profiles collection
A Change Stream is a server-side cursor that tails the MongoDB oplog and pushes change events to your application. Because your MongoDB instance is configured as a replica set, Change Streams are already available.
- In IntelliJ IDEA, create a new package stream inside src/main/kotlin/com/nextwork/syncengine/.
- Create a new Kotlin file called ChangeStreamListener.kt inside the stream package and paste the following code:
package com.nextwork.syncengine.stream
import com.mongodb.client.model.Aggregates
import com.mongodb.client.model.Filters
import com.mongodb.client.model.changestream.FullDocument
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.cache.ProfileCache
import com.nextwork.syncengine.model.SharedProfile
import kotlinx.coroutines.*
import org.bson.BsonDocument
class ChangeStreamListener(
private val db: MongoDatabase,
private val cache: ProfileCache
) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
@Volatile private var resumeToken: BsonDocument? = null
fun start() {
scope.launch {
while (isActive) {
try {
listenForChanges()
} catch (e: Exception) {
// On stream invalidation, clear token and do full re-sync
resumeToken = null
delay(1_000L)
}
}
}
}
private suspend fun listenForChanges() {
val pipeline = listOf(
Aggregates.match(
Filters.`in`("operationType", listOf("insert", "update", "replace"))
)
)
val stream = if (resumeToken != null) {
profiles.watch(pipeline).resumeAfter(resumeToken!!).fullDocument(FullDocument.UPDATE_LOOKUP)
} else {
profiles.watch(pipeline).fullDocument(FullDocument.UPDATE_LOOKUP)
}
stream.collect { event ->
resumeToken = event.resumeToken
val incoming = event.fullDocument ?: return@collect
val cached = cache.getProfile(incoming.id)
if (incoming.contentHash != cached.contentHash) {
cache.updateCached(incoming)
}
}
}
fun shutdown() {
scope.cancel()
}
}What does this code do?
The ChangeStreamListener class opens a MongoDB Change Stream that monitors the profiles collection for real-time changes. It combines three critical patterns in one class.
- An aggregation pipeline filters events to only insert, update, and replace operations. Delete events are ignored because profile deletion follows a separate lifecycle.
- The fullDocument(FullDocument.UPDATE_LOOKUP) option tells MongoDB to return the full current document on updates, not just the changed fields.
- A while (isActive) loop with try/catch ensures the stream automatically reconnects if it disconnects.
- Save ChangeStreamListener.kt.
Seeing unresolved references?
Make sure your import for FullDocument comes from com.mongodb.client.model.changestream.FullDocument, not from another package. Also confirm BsonDocument is imported from org.bson.BsonDocument.
Help me debug unresolved references in my ChangeStreamListener
Understand the hash-based echo filtering
When your server flushes deltas to MongoDB, the Change Stream on that same server fires with the change event. Without filtering, you would invalidate your own cache and trigger an unnecessary reload from the database.
- Find the collect block inside listenForChanges():
stream.collect { event ->
resumeToken = event.resumeToken
val incoming = event.fullDocument ?: return@collect
val cached = cache.getProfile(incoming.id)
if (incoming.contentHash != cached.contentHash) {
cache.updateCached(incoming)
}
}Why compare content hashes?
The contentHash field in SharedProfile acts as a fingerprint for the document's meaningful state. When your own server flushes, the resulting document in MongoDB has the same hash your local cache already holds. So the comparison incoming.contentHash != cached.contentHash evaluates to false and you skip the update.
When a different server writes, the incoming hash differs from yours. The listener replaces the stale cached snapshot with the fresh document from MongoDB.
Understand resume token resilience
Change Streams can disconnect due to network hiccups, MongoDB elections, or server restarts. Without a resume strategy, your listener would miss every event that occurred during the downtime.
- Find the start() method and the resumeToken property:
@Volatile private var resumeToken: BsonDocument? = null
fun start() {
scope.launch {
while (isActive) {
try {
listenForChanges()
} catch (e: Exception) {
// On stream invalidation, clear token and do full re-sync
resumeToken = null
delay(1_000L)
}
}
}
}How does the resume token work?
Every change event includes a resumeToken that marks the stream's position in the oplog. The listener stores this token after processing each event.
On reconnection, resumeAfter(resumeToken) tells MongoDB to replay events from where you left off. No events are missed during the disconnect window.
If the stream is fully invalidated (the token is too old or the collection was dropped), the catch block clears the token and waits 1 second before retrying. The next listenForChanges() call opens a fresh stream from the current point. The cache will refresh naturally on the next profile access via Caffeine's loader.
💡 Why use @Volatile on resumeToken?
The @Volatile annotation guarantees that reads and writes to resumeToken are visible across threads immediately. The coroutine collecting events and the outer retry loop may execute on different threads from the Dispatchers.IO pool.
✔️ Awesome, I've got everything!
Great. Double check you have saved ChangeStreamListener.kt and that IntelliJ shows no red underlines.
ⓧ I'd like to double check the full code
Here is the complete ChangeStreamListener.kt file:
package com.nextwork.syncengine.stream
import com.mongodb.client.model.Aggregates
import com.mongodb.client.model.Filters
import com.mongodb.client.model.changestream.FullDocument
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.cache.ProfileCache
import com.nextwork.syncengine.model.SharedProfile
import kotlinx.coroutines.*
import org.bson.BsonDocument
class ChangeStreamListener(
private val db: MongoDatabase,
private val cache: ProfileCache
) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
@Volatile private var resumeToken: BsonDocument? = null
fun start() {
scope.launch {
while (isActive) {
try {
listenForChanges()
} catch (e: Exception) {
// On stream invalidation, clear token and do full re-sync
resumeToken = null
delay(1_000L)
}
}
}
}
private suspend fun listenForChanges() {
val pipeline = listOf(
Aggregates.match(
Filters.`in`("operationType", listOf("insert", "update", "replace"))
)
)
val stream = if (resumeToken != null) {
profiles.watch(pipeline).resumeAfter(resumeToken!!).fullDocument(FullDocument.UPDATE_LOOKUP)
} else {
profiles.watch(pipeline).fullDocument(FullDocument.UPDATE_LOOKUP)
}
stream.collect { event ->
resumeToken = event.resumeToken
val incoming = event.fullDocument ?: return@collect
val cached = cache.getProfile(incoming.id)
if (incoming.contentHash != cached.contentHash) {
cache.updateCached(incoming)
}
}
}
fun shutdown() {
scope.cancel()
}
}Your sync engine now listens for cross-server changes in real time. Next up, you will extend the session system to support coop profiles where multiple players share a single profile across different servers.
Add Coop Profile Merging
Your Change Stream listener now keeps every server's cache consistent in real time. The sync engine handles solo players perfectly. But what about coop profiles that multiple players share across different servers?
In this step, you will extend the session system to support shared profiles. Multiple servers will hold "partial" sessions for the same profile, writing deltas concurrently without conflicts. The key insight is that $inc operations are commutative. Two servers incrementing the same counter simultaneously both succeed with correct totals.
In this step, get ready to:
- Extend the SessionManager to support partial sessions for coop profiles.
- Implement a field-access control layer that restricts session-owned writes.
- Wire the Change Stream to refresh coop members' cached views automatically.
Extend the SessionManager for shared sessions
The current SessionManager uses exclusive ownership. One server claims a profile and no other server can touch it. For coop profiles, multiple servers need simultaneous access.
Why partial sessions work for deltas
Delta-accumulator fields (currency, playtime, blocks broken) use MongoDB's $inc operator, which is atomic and commutative. Two servers both issuing $inc on the same field simultaneously will both succeed with correct totals.
Session-owned fields (inventory, position) are NOT commutative. Two servers setting different inventory states would lose one set of changes. These require exclusive "full" session ownership.
- In SessionManager.kt, find the package declaration at the top of the file. Replace the entire file contents with the following code that adds session type tracking:
package com.nextwork.syncengine.session
import com.mongodb.client.model.Filters
import com.mongodb.client.model.FindOneAndUpdateOptions
import com.mongodb.client.model.IndexOptions
import com.mongodb.client.model.ReturnDocument
import com.mongodb.client.model.Updates
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import kotlinx.coroutines.*
import org.bson.Document
import java.util.concurrent.TimeUnit
// Distinguishes full ownership from partial coop access
enum class SessionType { FULL, PARTIAL }
// Holds session metadata for diagnostics and queries
data class SessionInfo(val serverId: String, val type: SessionType, val expiresAt: Long)
class SessionManager(
private val db: MongoDatabase,
private val serverId: String
) {
private val sessions = db.getCollection<Document>("sessions")
private val heartbeatJobs = mutableMapOf<String, Job>()
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
// Tracks which session type this server holds for each profile
private val localSessions = mutableMapOf<String, SessionType>()
suspend fun init() {
sessions.createIndex(
Document("expiresAt", 1),
IndexOptions().expireAfter(0, TimeUnit.SECONDS)
)
}
suspend fun claimSession(profileId: String): Boolean {
val now = System.currentTimeMillis()
val expiry = now + 60_000L
val result = sessions.findOneAndUpdate(
Filters.and(
Filters.eq("_id", profileId),
Filters.or(
Filters.exists("_id", false),
Filters.lt("expiresAt", now)
)
),
Updates.combine(
Updates.set("serverId", serverId),
Updates.set("type", "FULL"),
Updates.set("claimedAt", now),
Updates.set("expiresAt", expiry)
),
FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER)
) ?: return false
val claimed = result.getString("serverId") == serverId
if (claimed) localSessions[profileId] = SessionType.FULL
startHeartbeat(profileId)
return claimed
}What does this code do?
The SessionType enum creates two distinct access levels. FULL means exclusive ownership (can write any field). PARTIAL means coop delta-only access.
The localSessions map tracks which session type this server currently holds for each profile. The modified claimSession now writes "FULL" to MongoDB and stores the type locally.
- Directly below the claimSession function, add the coop session methods and the modified releaseSession:
// Partial sessions are always allowed for coop delta accumulation
suspend fun claimPartialSession(profileId: String): Boolean {
localSessions[profileId] = SessionType.PARTIAL
startHeartbeat(profileId)
return true
}
// Query what type of session this server holds
fun getSessionType(profileId: String): SessionType? = localSessions[profileId]
// Only full session holders can write inventory/position
fun canWriteSessionOwned(profileId: String): Boolean =
localSessions[profileId] == SessionType.FULL
suspend fun releaseSession(profileId: String) {
heartbeatJobs[profileId]?.cancel()
heartbeatJobs.remove(profileId)
localSessions.remove(profileId)
sessions.deleteOne(Filters.and(
Filters.eq("_id", profileId),
Filters.eq("serverId", serverId)
))
}
private fun startHeartbeat(profileId: String) {
heartbeatJobs[profileId] = scope.launch {
while (isActive) {
delay(30_000L)
val expiry = System.currentTimeMillis() + 60_000L
sessions.updateOne(
Filters.and(
Filters.eq("_id", profileId),
Filters.eq("serverId", serverId)
),
Updates.set("expiresAt", expiry)
)
}
}
}
fun shutdown() {
scope.cancel()
}
}What does this code do?
- claimPartialSession() always succeeds. It stores the profile as PARTIAL locally and starts a heartbeat. No exclusive lock is needed because partial sessions only write commutative deltas.
- canWriteSessionOwned() is the gate that enforces field access. Only servers holding a FULL session can modify inventory or position.
- releaseSession() now cleans up the localSessions map alongside the heartbeat and MongoDB document.
- Save SessionManager.kt.
- Verify in IntelliJ that the file compiles without errors. You should see no red underlines in the editor gutter.
Seeing unresolved reference errors?
- Make sure the SessionType enum and SessionInfo data class are defined outside (above) the SessionManager class, not inside it.
- Confirm localSessions is declared as a class property (inside the class body, alongside heartbeatJobs).
Help me debug my SessionManager compilation errors
✔️ Awesome, I've got everything!
Great. Double check you have saved SessionManager.kt before moving on.
ⓧ I'd like to double check the full code
package com.nextwork.syncengine.session
import com.mongodb.client.model.Filters
import com.mongodb.client.model.FindOneAndUpdateOptions
import com.mongodb.client.model.IndexOptions
import com.mongodb.client.model.ReturnDocument
import com.mongodb.client.model.Updates
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import kotlinx.coroutines.*
import org.bson.Document
import java.util.concurrent.TimeUnit
enum class SessionType { FULL, PARTIAL }
data class SessionInfo(val serverId: String, val type: SessionType, val expiresAt: Long)
class SessionManager(
private val db: MongoDatabase,
private val serverId: String
) {
private val sessions = db.getCollection<Document>("sessions")
private val heartbeatJobs = mutableMapOf<String, Job>()
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val localSessions = mutableMapOf<String, SessionType>()
suspend fun init() {
sessions.createIndex(
Document("expiresAt", 1),
IndexOptions().expireAfter(0, TimeUnit.SECONDS)
)
}
suspend fun claimSession(profileId: String): Boolean {
val now = System.currentTimeMillis()
val expiry = now + 60_000L
val result = sessions.findOneAndUpdate(
Filters.and(
Filters.eq("_id", profileId),
Filters.or(
Filters.exists("_id", false),
Filters.lt("expiresAt", now)
)
),
Updates.combine(
Updates.set("serverId", serverId),
Updates.set("type", "FULL"),
Updates.set("claimedAt", now),
Updates.set("expiresAt", expiry)
),
FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER)
) ?: return false
val claimed = result.getString("serverId") == serverId
if (claimed) localSessions[profileId] = SessionType.FULL
startHeartbeat(profileId)
return claimed
}
suspend fun claimPartialSession(profileId: String): Boolean {
localSessions[profileId] = SessionType.PARTIAL
startHeartbeat(profileId)
return true
}
fun getSessionType(profileId: String): SessionType? = localSessions[profileId]
fun canWriteSessionOwned(profileId: String): Boolean =
localSessions[profileId] == SessionType.FULL
suspend fun releaseSession(profileId: String) {
heartbeatJobs[profileId]?.cancel()
heartbeatJobs.remove(profileId)
localSessions.remove(profileId)
sessions.deleteOne(Filters.and(
Filters.eq("_id", profileId),
Filters.eq("serverId", serverId)
))
}
private fun startHeartbeat(profileId: String) {
heartbeatJobs[profileId] = scope.launch {
while (isActive) {
delay(30_000L)
val expiry = System.currentTimeMillis() + 60_000L
sessions.updateOne(
Filters.and(
Filters.eq("_id", profileId),
Filters.eq("serverId", serverId)
),
Updates.set("expiresAt", expiry)
)
}
}
}
fun shutdown() {
scope.cancel()
}
}Implement the field-access control layer
The SessionManager now knows who holds what type of session. The ProfileCache needs to enforce this. Partial sessions can call addDelta() freely, but only full session holders can write session-owned fields like inventory and position.
- Open ProfileCache.kt. Replace the entire file contents with this updated version that adds the SessionManager dependency and field-access control:
package com.nextwork.syncengine.cache
import com.github.benmanes.caffeine.cache.Caffeine
import com.mongodb.client.model.Filters
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.model.SharedProfile
import com.nextwork.syncengine.session.SessionManager
import kotlinx.coroutines.runBlocking
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
class ProfileCache(
private val db: MongoDatabase,
private val sessionManager: SessionManager
) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val accumulators = ConcurrentHashMap<String, DeltaAccumulator>()
private val cache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterAccess(10, TimeUnit.MINUTES)
.build<String, SharedProfile> { id ->
runBlocking {
profiles.find(Filters.eq("_id", id)).firstOrNull()
?: SharedProfile(id = id)
}
}
fun getAccumulator(profileId: String): DeltaAccumulator =
accumulators.getOrPut(profileId) { DeltaAccumulator() }
fun addDelta(profileId: String, field: String, amount: Long) {
val acc = getAccumulator(profileId)
when (field) {
"currency" -> acc.currency.addAndGet(amount)
"playtime" -> acc.playtime.addAndGet(amount)
"blocksPlaced" -> acc.blocksPlaced.addAndGet(amount)
"blocksBroken" -> acc.blocksBroken.addAndGet(amount)
"deaths" -> acc.deaths.addAndGet(amount)
}
}
// Gate for session-owned fields: rejects writes from partial sessions
fun setSessionOwned(profileId: String, field: String, value: Any): Boolean {
if (!sessionManager.canWriteSessionOwned(profileId)) return false
val current = cache.getIfPresent(profileId) ?: return false
val updated = when (field) {
"inventory" -> current.copy(inventory = value as String)
"position" -> current.copy(position = value as String)
else -> return false
}
cache.put(profileId, updated)
return true
}
fun getProfile(profileId: String): SharedProfile {
val base = cache.get(profileId)!!
val acc = accumulators[profileId] ?: return base
return base.copy(
currency = base.currency + acc.currency.get(),
playtime = base.playtime + acc.playtime.get(),
blocksPlaced = base.blocksPlaced + acc.blocksPlaced.get(),
blocksBroken = base.blocksBroken + acc.blocksBroken.get(),
deaths = base.deaths + acc.deaths.get()
)
}
fun updateCached(profile: SharedProfile) {
cache.put(profile.id, profile)
}
fun getAllDirtyAccumulators(): Map<String, DeltaAccumulator> =
accumulators.filter { it.value.isDirty() }
}What does this code do?
- The constructor now accepts a SessionManager reference. This lets the cache check session ownership before allowing writes.
- setSessionOwned() is the new gatekeeper. It calls canWriteSessionOwned() on the SessionManager. If the server only holds a PARTIAL session, the write is rejected (returns false).
- addDelta() has no ownership check. Any server with any session type can accumulate deltas, because $inc is conflict-free.
- Save ProfileCache.kt.
- IntelliJ will show a compilation error in FlushScheduler.kt because ProfileCache's constructor signature changed. This is expected. You will need to update how you construct ProfileCache wherever it is instantiated (passing both the database and the SessionManager). For now, confirm that ProfileCache.kt itself compiles cleanly.
Seeing type mismatch errors?
- Check that the import for SessionManager is present: import com.nextwork.syncengine.session.SessionManager.
- Confirm the constructor parameter order matches: db first, sessionManager second.
Help me fix my ProfileCache compilation errors
✔️ Awesome, I've got everything!
Great. Double check you have saved ProfileCache.kt before moving on.
ⓧ I'd like to double check the full code
package com.nextwork.syncengine.cache
import com.github.benmanes.caffeine.cache.Caffeine
import com.mongodb.client.model.Filters
import com.mongodb.kotlin.client.coroutine.MongoDatabase
import com.nextwork.syncengine.model.SharedProfile
import com.nextwork.syncengine.session.SessionManager
import kotlinx.coroutines.runBlocking
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
class ProfileCache(
private val db: MongoDatabase,
private val sessionManager: SessionManager
) {
private val profiles = db.getCollection<SharedProfile>("profiles")
private val accumulators = ConcurrentHashMap<String, DeltaAccumulator>()
private val cache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterAccess(10, TimeUnit.MINUTES)
.build<String, SharedProfile> { id ->
runBlocking {
profiles.find(Filters.eq("_id", id)).firstOrNull()
?: SharedProfile(id = id)
}
}
fun getAccumulator(profileId: String): DeltaAccumulator =
accumulators.getOrPut(profileId) { DeltaAccumulator() }
fun addDelta(profileId: String, field: String, amount: Long) {
val acc = getAccumulator(profileId)
when (field) {
"currency" -> acc.currency.addAndGet(amount)
"playtime" -> acc.playtime.addAndGet(amount)
"blocksPlaced" -> acc.blocksPlaced.addAndGet(amount)
"blocksBroken" -> acc.blocksBroken.addAndGet(amount)
"deaths" -> acc.deaths.addAndGet(amount)
}
}
fun setSessionOwned(profileId: String, field: String, value: Any): Boolean {
if (!sessionManager.canWriteSessionOwned(profileId)) return false
val current = cache.getIfPresent(profileId) ?: return false
val updated = when (field) {
"inventory" -> current.copy(inventory = value as String)
"position" -> current.copy(position = value as String)
else -> return false
}
cache.put(profileId, updated)
return true
}
fun getProfile(profileId: String): SharedProfile {
val base = cache.get(profileId)!!
val acc = accumulators[profileId] ?: return base
return base.copy(
currency = base.currency + acc.currency.get(),
playtime = base.playtime + acc.playtime.get(),
blocksPlaced = base.blocksPlaced + acc.blocksPlaced.get(),
blocksBroken = base.blocksBroken + acc.blocksBroken.get(),
deaths = base.deaths + acc.deaths.get()
)
}
fun updateCached(profile: SharedProfile) {
cache.put(profile.id, profile)
}
fun getAllDirtyAccumulators(): Map<String, DeltaAccumulator> =
accumulators.filter { it.value.isDirty() }
}Wire the Change Stream to refresh coop views
The ChangeStreamListener you built in the previous step already does exactly what coop needs. No code changes are required here. Let's trace through the flow to see why.
How the existing Change Stream enables coop sync
Consider this scenario: Player A and Player B share a coop profile. Player A is on Server A, Player B is on Server B. Both servers hold partial sessions for the same profile.
- Player A earns 50 coins on Server A. Server A's FlushScheduler issues $inc on the shared profile.
- MongoDB applies the $inc. The Change Stream fires on all watching servers.
- Server A receives the event. The contentHash matches its own write (echo), so it skips the update.
- Server B receives the event. The contentHash differs from its cached version, so it calls updateCached() with the fresh document.
- Next time Player B reads the profile on Server B, getProfile() returns the updated snapshot with the 50 coins visible.
The hash-based echo filtering handles the "don't invalidate your own writes" problem. The updateCached() call handles the "show coop partners updated values" problem. Both were already built in Step 4.
Why $inc is the secret to conflict-free coop
If both servers flush deltas simultaneously, MongoDB handles them serially at the document level. Server A's $inc: 50 and Server B's $inc: 30 both succeed. The final value is correctly 80, regardless of which write arrived first.
This is why session-owned fields (inventory, position) are gated behind canWriteSessionOwned(). A $set operation is NOT commutative. Two competing $set calls would result in last-write-wins data loss.
Your coop profile system is now complete. Multiple servers can accumulate deltas for the same shared profile without conflicts, and the Change Stream keeps every coop member's view fresh. Next up, you will tackle the Secret Mission: optimistic retry logic for inventory transfers where $inc won't save you and real conflict resolution becomes necessary.
Secret mission
Optimistic Retry for Inventory Transfers
Your delta accumulators handle numeric fields with zero conflicts because $inc is commutative. But what about inventory transfers between coop members? Two players on different servers both moving the same diamond sword to the shared bank at the same time creates a duplication bug. Build a conflict resolution system with optimistic locking and exponential backoff to guarantee exactly-once transfers.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- MongoDB syncengine database (with profiles and sessions collections).
- MongoDB replica set configuration (enabled via rs.initiate()).
- Your SyncEngine project folder on the Desktop.
Keep everything running
No action needed. Choose this if you're still actively building on this plugin or plan to extend it further.
- Your MongoDB replica set stays running and ready for development.
- Your project files remain on disk, ready to open in IntelliJ IDEA.
- The syncengine database retains all test data you created during development.
Pause - I'll come back to this later
Stop MongoDB to free up system memory and CPU. Your data stays intact on disk for when you return.
- Stop the MongoDB service on macOS by running:
brew services stop mongodb-community- On Windows, stop the MongoDB service by running this in an Administrator terminal:
net stop MongoDB- Your project files and database data remain on disk. When you are ready to resume, start MongoDB again and open the project in IntelliJ IDEA.
Delete - I don't want to use this again
Remove all project resources and start fresh. This permanently deletes your code and test data.
- Drop the syncengine database by opening mongosh and running:
use syncengine
db.dropDatabase()- Remove the project folder on macOS or Linux by running:
cd ~/Desktop
rm -rf syncengine- On Windows, remove the project folder by running:
cd ~\Desktop
rmdir /s /q syncengine- If you no longer need the replica set configuration but want to keep MongoDB installed, you can restart MongoDB as a standalone instance by removing the --replSet flag from your MongoDB configuration file.
Nice Work!
Nice Work!
You just built a production-grade cross-server data synchronization engine for Minecraft. This plugin handles 1000+ concurrent players across multiple servers with minimal database load.
Here's what you learned:
- Built a write-back cache with delta accumulators that reduces database writes by 99.5%, flushing batched $inc operations every 5 seconds instead of writing on every player action.
- Built MongoDB Change Stream listeners that invalidate cached data across servers in real time, using hash-based echo filtering to skip your own writes and resume tokens for resilient reconnection.
- Built a session ownership system with TTL heartbeats and coop profile support that allows multiple servers to safely merge conflict-free deltas on shared profiles.
- Secret Mission: Implemented optimistic retry logic with exponential backoff for safe inventory transfers under concurrent modification.
Ready to quiz yourself?