diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 113e77d8..60d82ce5 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -220,6 +220,8 @@ node = create_node("mesh.cc", runtime_id="python-runtime") node.register_document(note_doc) node.put(note_doc, "note:1", {"body": "hello"}) live_note = node.get_required(note_doc, "note:1") +put_message = node.put_raw_message(note_doc, "note:2", {"body": "wire me"}, shard_id="interop", shard_epoch=1) +delete_message = node.delete_raw_message(note_doc, "note:2", shard_id="interop", shard_epoch=1) peers = CultMeshPeerCatalog() response = peers.create_response(peer_exchange_request("pex-1", verse_id="local")) diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index a1e0026b..fa76cecd 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -4,9 +4,17 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Any -from cultcache_py import CultCache, SingleFileMessagePackBackingStore +from cultcache_py import CultCache, CultCacheEnvelope, SingleFileMessagePackBackingStore from cultcache_py.documents import DocumentDefinition -from cultnet_py import CultNetAppliedRecord, CultNetRawClient, apply_raw_snapshot, apply_shard_log_response +from cultnet_py import ( + CultNetAppliedRecord, + CultNetMessage, + CultNetRawClient, + apply_raw_snapshot, + apply_shard_log_response, + document_delete, + document_put_raw, +) @dataclass @@ -22,6 +30,36 @@ class CultMeshNode: def put(self, document: DocumentDefinition[Any], key: str, value: Any) -> None: self.cache.put(document, key, value) + def put_raw_message( + self, + document: DocumentDefinition[Any], + key: str, + value: Any, + *, + message_id: str = "", + shard_id: str | None = None, + shard_epoch: int | None = None, + ) -> CultNetMessage: + catalog_entry = document.catalog_entry() + envelope = CultCacheEnvelope.create( + key=key, + type=document.type, + payload=document.encode_payload(value), + schema_id=catalog_entry.schema_id, + catalog_entry=catalog_entry, + ) + self.cache.put_envelope(document, envelope) + return document_put_raw( + message_id=message_id, + key=key, + schema_id=envelope.schema_id or catalog_entry.schema_id, + stored_at=envelope.stored_at, + payload=envelope.payload, + source_runtime_id=self.runtime_id, + shard_id=shard_id, + shard_epoch=shard_epoch, + ) + def get(self, document: DocumentDefinition[Any], key: str) -> Any: return self.cache.get(document, key) @@ -31,6 +69,25 @@ class CultMeshNode: def delete(self, document: DocumentDefinition[Any], key: str) -> None: self.cache.delete(document, key) + def delete_raw_message( + self, + document: DocumentDefinition[Any], + key: str, + *, + message_id: str = "", + shard_id: str | None = None, + shard_epoch: int | None = None, + ) -> CultNetMessage: + schema_id = document.catalog_entry().schema_id + self.delete(document, key) + return document_delete( + message_id=message_id, + schema_id=schema_id, + record_key=key, + shard_id=shard_id, + shard_epoch=shard_epoch, + ) + def snapshot(self) -> dict[str, dict[str, Any]]: return self.cache.snapshot() diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index c0ece8cf..90dec6f0 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -517,6 +517,51 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(log_changes[0].change_kind, "updated") self.assertEqual(node.get_required(document, "item:node").value, 9) + def test_cultmesh_node_emits_raw_put_and_delete_messages_for_local_writes(self) -> None: + document = define_database_entry_type( + "mesh.emit_item", + [ + ("name", 0), + ("category", 1), + ("value", 2, 0), + ], + cls=Item, + ) + node = create_node(runtime_id="python-node") + node.register_document(document) + + put = node.put_raw_message( + document, + "item:emit", + Item("wand", "gear", 4), + message_id="put-emit", + shard_id="mesh", + shard_epoch=1, + ).to_wire() + self.assertEqual(node.get_required(document, "item:emit").name, "wand") + self.assertEqual(put["schemaVersion"], "cultnet.document_put_raw.v0") + self.assertEqual(put["messageId"], "put-emit") + self.assertEqual(put["document"]["schemaId"], document.catalog_entry().schema_id) + self.assertEqual(put["document"]["recordKey"], "item:emit") + self.assertEqual(put["document"]["sourceRuntimeId"], "python-node") + self.assertEqual(put["shardId"], "mesh") + self.assertEqual(put["shardEpoch"], 1) + + delete = node.delete_raw_message( + document, + "item:emit", + message_id="delete-emit", + shard_id="mesh", + shard_epoch=1, + ).to_wire() + self.assertIsNone(node.get(document, "item:emit")) + self.assertEqual(delete["schemaVersion"], "cultnet.document_delete.v0") + self.assertEqual(delete["messageId"], "delete-emit") + self.assertEqual(delete["schemaId"], document.catalog_entry().schema_id) + self.assertEqual(delete["recordKey"], "item:emit") + self.assertEqual(delete["shardId"], "mesh") + self.assertEqual(delete["shardEpoch"], 1) + def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None: message = document_delete( message_id="delete-1", diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 228d8223..adf57fbf 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -306,6 +306,11 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state assert.equal(pythonCultMeshNodeSync.synced[0].recordKey, "note:python-peer"); assert.equal(pythonCultMeshNodeSync.note.authorRuntimeId, "python-peer"); + const pythonCultMeshNodeEmit = await runPythonCultMeshNodeEmit(pythonPort); + assert.equal(pythonCultMeshNodeEmit.synced[0].recordKey, "note:python-node-emit"); + assert.equal(pythonCultMeshNodeEmit.note.authorRuntimeId, "python-node-emit"); + assert.equal(pythonCultMeshNodeEmit.note.title, "Python node emitted raw put"); + const pythonChange = await requestPythonDatabaseChange(pythonPort, { schemaVersion: "cultnet.database_subscribe.v0", messageId: "python-subscribe", @@ -802,6 +807,71 @@ async function runPythonCultMeshNodeSync(port: number): Promise { return JSON.parse(stdout); } +async function runPythonCultMeshNodeEmit(port: number): Promise { + const schemaPathJson = JSON.stringify(interopSchemaPath); + const script = [ + "import json", + "import socket", + "import time", + "from pathlib import Path", + "from cultcache_py import define_database_entry_type", + "from cultmesh_py import create_node", + "from cultnet_py import CultNetRawClient, write_frame", + `schema_json = Path(${schemaPathJson}).read_text(encoding='utf-8')`, + "note_doc = define_database_entry_type(", + " 'cultnet.interop-note',", + " [('schemaVersion', 0), ('documentId', 1), ('authorRuntimeId', 2), ('title', 3), ('body', 4), ('tags', 5, [])],", + ` schema_id='${interopNoteSchemaId}',`, + " schema_version='cultnet.interop_note.v0',", + " canonical_schema_json=schema_json,", + ")", + "writer = create_node(runtime_id='python-node-emit')", + "writer.register_document(note_doc)", + "message = writer.put_raw_message(", + " note_doc,", + " 'note:python-node-emit',", + " {", + " 'schemaVersion': 'cultnet.interop_note.v0',", + " 'documentId': 'note:python-node-emit',", + " 'authorRuntimeId': 'python-node-emit',", + " 'title': 'Python node emitted raw put',", + " 'body': 'node-level Python emit reached the peer',", + " 'tags': ['python', 'node', 'emit'],", + " },", + " message_id='python-node-emit-put',", + " shard_id='interop',", + " shard_epoch=1,", + ")", + `with socket.create_connection(('127.0.0.1', ${port}), timeout=4.0) as connection:`, + " stream = connection.makefile('rwb')", + " write_frame(stream, message.to_bytes())", + " stream.flush()", + "reader = create_node(runtime_id='python-node-emit-reader')", + "reader.register_document(note_doc)", + `client = CultNetRawClient('127.0.0.1', ${port}, timeout_seconds=4.0)`, + "changes = []", + "for _ in range(20):", + ` changes = reader.sync_snapshot(client, schema_ids=['${interopNoteSchemaId}'], record_keys=['note:python-node-emit'])`, + " if changes:", + " break", + " time.sleep(0.1)", + "note = reader.get_required(note_doc, 'note:python-node-emit')", + "print(json.dumps({", + " 'synced': [{'schemaId': change.schema_id, 'recordKey': change.record_key} for change in changes],", + " 'note': note,", + "}))", + ].join("\n"); + const { stdout } = await execFileAsync(pythonCommand, ["-c", script], { + cwd: cultcachePyRoot, + env: { + ...process.env, + PYTHONPATH: cultcachePySrc, + }, + timeout: 8000, + }); + return JSON.parse(stdout); +} + async function requestPythonDatabaseChange(port: number, subscribe: Record, put: Record): Promise { const socket = connectTcp(port, "127.0.0.1"); const framer = new LengthPrefixedMessageFramer();