diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index 4b7bd68a..32567bef 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt @@ -225,6 +225,34 @@ fun bytesDocument( global: Boolean = false, ): CultDocumentDefinition = cultDocument(ByteArrayDocumentCodec(documentType, schemaVersion), global) +private data class KotlinCanonicalNote( + val schemaVersion: String, + val body: String, +) + +private data class KotlinUiNote( + val schemaVersion: String, + val body: String, +) + +private class KotlinAliasNoteCodec( + override val documentType: String, + override val schemaVersion: String, + private val create: (String, String) -> T, + private val readSchemaVersion: (T) -> String, + private val readBody: (T) -> String, +) : CultDocumentCodec { + override fun encode(value: T): ByteArray = + MessagePackWriter() + .value(listOf(readSchemaVersion(value), readBody(value))) + .toByteArray() + + override fun decode(payload: ByteArray): T { + val slots = anyList(MessagePackReader(payload).readAny()) + return create(slots[0] as String, slots[1] as String) + } +} + class CultMeshNode( val cache: CultCache = CultCache(), private val random: SecureRandom = SecureRandom(), @@ -270,6 +298,9 @@ class CultMeshNode( fun applyRawSnapshotResponse(response: CultNetMessage): List = cache.applyRawSnapshotResponse(response) + fun syncDocument(response: CultNetMessage, document: CultDocumentDefinition, key: String): T = + cache.syncDocument(response, document, key) + fun applyDocumentDelete(message: CultNetMessage): Boolean = cache.applyDocumentDelete(message) fun applyShardLogResponse(response: CultNetMessage): List = cache.applyShardLogResponse(response) @@ -1973,6 +2004,22 @@ fun CultCache.applyRawSnapshotResponse(response: CultNetMessage): List { .map { applyRawDocumentRecord(it) } } +fun CultCache.syncDocument( + response: CultNetMessage, + document: CultDocumentDefinition, + key: String, +): T { + require(response.schemaVersion == "cultnet.snapshot_response_raw.v0") { + "Expected cultnet.snapshot_response_raw.v0, received ${response.schemaVersion}" + } + val rawDocument = mapList(response.body["documents"]) + .map { rawDocumentRecordFromWire(it) } + .firstOrNull { it.schemaId == document.schemaVersion && it.recordKey == key } + ?: throw NoSuchElementException("No ${document.schemaVersion} document for key $key") + applyRawDocumentRecord(rawDocument) + return getRequired(document, key) +} + fun CultCache.applyDocumentDelete(message: CultNetMessage): Boolean { require(message.schemaVersion == "cultnet.document_delete.v0") { "Expected cultnet.document_delete.v0, received ${message.schemaVersion}" @@ -3085,6 +3132,36 @@ private fun cultCacheRawSnapshotsRoundTripThroughCultNetMessages() { check(target.require(notes, "note:2") == "second") check(target.recallGlobal(settings) == null) + val canonicalNote = cultDocument(KotlinAliasNoteCodec( + documentType = "kotlin.alias_note", + schemaVersion = "kotlin.alias_note.v1", + create = ::KotlinCanonicalNote, + readSchemaVersion = KotlinCanonicalNote::schemaVersion, + readBody = KotlinCanonicalNote::body, + )) + val uiNote = cultDocument(KotlinAliasNoteCodec( + documentType = "kotlin.alias_note.ui", + schemaVersion = "kotlin.alias_note.v1", + create = ::KotlinUiNote, + readSchemaVersion = KotlinUiNote::schemaVersion, + readBody = KotlinUiNote::body, + )) + source.cache.register(canonicalNote) + target.cache.register(uiNote) + source.remember(canonicalNote, "note:alias", KotlinCanonicalNote("kotlin.alias_note.v1", "canonical-to-ui")) + val aliasSnapshotResponse = source.createRawSnapshotResponse( + cultNetSnapshotRequest( + messageId = "snapshot-alias-note", + schemaIds = listOf(uiNote.schemaVersion), + recordKeys = listOf("note:alias"), + ), + storedAt = "2026-06-15T00:00:02Z", + sourceRuntimeId = "kotlin-source", + ) + val syncedAlias = target.syncDocument(aliasSnapshotResponse, uiNote, "note:alias") + check(syncedAlias == KotlinUiNote("kotlin.alias_note.v1", "canonical-to-ui")) + check(target.require(uiNote, "note:alias") == syncedAlias) + val rawPut = parseCultNetMessage( cultNetDocumentPutRaw( messageId = "put-note", diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 2803db51..39ea2e39 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -149,6 +149,20 @@ struct GhostlightAgentStateFixture { display_name: String, } +#[derive(Clone, Debug, PartialEq, DatabaseEntry)] +#[cultcache( + type = "ghostlight.agent-state.ui", + schema = "GhostlightAgentStateUiFixture" +)] +struct GhostlightAgentStateUiFixture { + #[cultcache(key = 0)] + schema_version: String, + #[cultcache(key = 1)] + agent_id: String, + #[cultcache(key = 2)] + display_name: String, +} + #[test] fn security_helpers_round_trip_encrypted_strings_and_validate_sessions() -> Result<()> { let server_security = CultNetServerSecurityOptions::development(); @@ -1682,6 +1696,66 @@ fn raw_snapshot_replication_preserves_messagepack_payload_bytes() -> Result<()> Ok(()) } +#[test] +fn raw_snapshot_replication_hydrates_same_schema_rust_aliases() -> Result<()> { + let temp = tempfile::tempdir()?; + let origin_store = temp.path().join("origin-alias.msgpack"); + let target_store = temp.path().join("target-alias.msgpack"); + let canonical = GhostlightAgentStateFixture { + schema_version: "ghostlight.agent_state.v0".to_string(), + agent_id: "epiphany.persona".to_string(), + display_name: "Persona".to_string(), + }; + + let mut origin_registry = CultNetDocumentRegistry::new(); + origin_registry.register(CultNetDocumentBinding::for_entry_with_schema_id::< + GhostlightAgentStateFixture, + >( + "ghostlight.agent_state.v0".to_string(), + "ghostlight.agent_state.v0".to_string(), + )); + let mut origin = CultCache::new(); + origin.register_entry_type::()?; + origin.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&origin_store)); + origin.pull_all_backing_stores()?; + origin.put("epiphany.persona", &canonical)?; + let raw_snapshot = origin_registry.create_raw_snapshot_response( + &origin, + "raw-snapshot-alias", + Some(&["ghostlight.agent_state.v0".to_string()]), + Some(&["epiphany.persona".to_string()]), + )?; + + let mut alias_registry = CultNetDocumentRegistry::new(); + alias_registry.register(CultNetDocumentBinding::for_entry_with_schema_id::< + GhostlightAgentStateUiFixture, + >( + "ghostlight.agent_state.v0".to_string(), + "ghostlight.agent_state.v0".to_string(), + )); + let mut target = CultCache::new(); + target.register_entry_type::()?; + target.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&target_store)); + target.pull_all_backing_stores()?; + + let applied = alias_registry + .apply_raw_snapshot_response::(&mut target, &raw_snapshot)?; + + assert_eq!( + applied, + vec![GhostlightAgentStateUiFixture { + schema_version: "ghostlight.agent_state.v0".to_string(), + agent_id: "epiphany.persona".to_string(), + display_name: "Persona".to_string(), + }] + ); + assert_eq!( + target.get_required::("epiphany.persona")?, + applied[0] + ); + Ok(()) +} + #[test] fn client_security_keeps_the_connection_key_visible_without_exposing_cipher_logic() -> Result<()> { let options = CultNetClientSecurityOptions::development();