diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 4eddde16..9db9fbcc 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -147,6 +147,8 @@ member order. ```python from cultnet_py import ( CultNetRawClient, + apply_raw_snapshot, + apply_shard_log_response, compute_simulation_claim_hash, database_subscribe, decode_frame, @@ -165,9 +167,13 @@ client = CultNetRawClient("127.0.0.1", 3075) catalog = client.fetch_schema_catalog(kinds=["wire_message"]) snapshot = client.fetch_snapshot(schema_ids=["cultnet.interop-note"]) shard_catalog = client.fetch_shard_catalog(schema_ids=["cultnet.interop-note"]) +# With a local CultCache and the matching registered document definitions: +# applied = apply_raw_snapshot(cache, [note_doc], snapshot) subscription = database_subscribe(subscription_id="ui", schema_ids=["cultnet.interop-note"]) shard_request = shard_catalog_request(message_id="shards", schema_ids=["cultnet.interop-note"]) +shard_log = client.fetch_shard_log(shard_id="interop", shard_epoch=1, after_sequence=0) +# log_changes = apply_shard_log_response(cache, [note_doc], shard_log) claim_hash = compute_simulation_claim_hash("frame:42", "subject:player-1", "hit") observation = simulation_observation( message_id="obs-1", @@ -249,8 +255,8 @@ Current receipts: snapshot, and shard catalog. - `packages/cultcache-py/tests/test_cultcache.py` covers Python CultMesh Verse catalog, peer exchange, authority lease, stream negotiation, CultNet - helper shapes, simulation claim hashing, and witness artifact bundle payload - slots. + helper shapes, raw snapshot/shard-log application, simulation claim hashing, + and witness artifact bundle payload slots. - `cultcache_py`, `cultnet_py`, and `cultmesh_py` ship `py.typed` markers so downstream type checkers can inspect the package surface instead of treating the runtime as an untyped xenos swamp. diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index 8508465a..6acfc65a 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -18,10 +18,21 @@ from .messages import ( snapshot_request, witness_artifact_bundle, ) +from .replication import ( + CultNetAppliedRecord, + apply_raw_document_record, + apply_raw_snapshot, + apply_shard_log_response, + schema_document_map, +) __all__ = [ "CultNetMessage", "CultNetRawClient", + "CultNetAppliedRecord", + "apply_raw_document_record", + "apply_raw_snapshot", + "apply_shard_log_response", "compute_simulation_claim_hash", "database_subscribe", "database_unsubscribe", @@ -35,6 +46,7 @@ __all__ = [ "parse_message", "read_frame", "schema_catalog_request", + "schema_document_map", "shard_catalog_request", "shard_log_request", "simulation_observation", diff --git a/packages/cultcache-py/src/cultnet_py/replication.py b/packages/cultcache-py/src/cultnet_py/replication.py new file mode 100644 index 00000000..a3d40d8f --- /dev/null +++ b/packages/cultcache-py/src/cultnet_py/replication.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Iterable + +from cultcache_py import CultCache, CultCacheEnvelope +from cultcache_py.documents import DocumentDefinition + + +@dataclass(frozen=True) +class CultNetAppliedRecord: + schema_id: str + record_key: str + change_kind: str + value: Any | None = None + + +def schema_document_map(documents: Iterable[DocumentDefinition[Any]]) -> dict[str, DocumentDefinition[Any]]: + mapped: dict[str, DocumentDefinition[Any]] = {} + for document in documents: + catalog_entry = document.catalog_entry() + mapped[catalog_entry.schema_id] = document + for compatible_schema_id in catalog_entry.compatible_schema_ids: + mapped[compatible_schema_id] = document + return mapped + + +def apply_raw_document_record( + cache: CultCache, + documents_by_schema_id: dict[str, DocumentDefinition[Any]], + record: dict[str, Any], +) -> CultNetAppliedRecord: + schema_id = str(record["schemaId"]) + document = documents_by_schema_id[schema_id] + envelope = CultCacheEnvelope( + key=str(record["recordKey"]), + type=document.type, + schema_id=schema_id, + payload=bytes(record["payload"]), + stored_at=str(record.get("storedAt") or ""), + catalog_entry=document.catalog_entry(), + ) + value = cache.put_envelope(document, envelope) + return CultNetAppliedRecord( + schema_id=schema_id, + record_key=envelope.key, + change_kind="added", + value=value, + ) + + +def apply_raw_snapshot( + cache: CultCache, + documents: Iterable[DocumentDefinition[Any]], + response: dict[str, Any], +) -> list[CultNetAppliedRecord]: + if response.get("schemaVersion") != "cultnet.snapshot_response_raw.v0": + raise ValueError(f"Expected cultnet.snapshot_response_raw.v0, received {response.get('schemaVersion')!r}") + documents_by_schema_id = schema_document_map(documents) + return [ + apply_raw_document_record(cache, documents_by_schema_id, record) + for record in response.get("documents", []) + if isinstance(record, dict) + ] + + +def apply_shard_log_response( + cache: CultCache, + documents: Iterable[DocumentDefinition[Any]], + response: dict[str, Any], +) -> list[CultNetAppliedRecord]: + if response.get("schemaVersion") != "cultnet.shard_log_response.v0": + raise ValueError(f"Expected cultnet.shard_log_response.v0, received {response.get('schemaVersion')!r}") + if response.get("resyncRequired") is True: + reason = response.get("reason") or "unspecified" + raise ValueError(f"Shard log response requires resync: {reason}") + documents_by_schema_id = schema_document_map(documents) + applied: list[CultNetAppliedRecord] = [] + for entry in response.get("entries", []): + if not isinstance(entry, dict): + continue + change_kind = entry.get("changeKind") + if change_kind in ("added", "updated") and isinstance(entry.get("put"), dict): + put = entry["put"] + result = apply_raw_document_record(cache, documents_by_schema_id, put["document"]) + applied.append(CultNetAppliedRecord(result.schema_id, result.record_key, str(change_kind), result.value)) + elif change_kind == "removed" and isinstance(entry.get("delete"), dict): + delete = entry["delete"] + schema_id = str(delete["schemaId"]) + record_key = str(delete["recordKey"]) + document = documents_by_schema_id[schema_id] + cache.delete(document, record_key) + applied.append(CultNetAppliedRecord(schema_id, record_key, "removed", None)) + return applied + diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index f4a363f3..a82f14ef 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -20,6 +20,8 @@ from cultcache_py.interop import read_note, write_note from cultnet_py import ( compute_simulation_claim_hash, CultNetRawClient, + apply_raw_snapshot, + apply_shard_log_response, database_subscribe, database_unsubscribe, decode_frame, @@ -331,6 +333,79 @@ class CultCacheTests(unittest.TestCase): "cultnet.shard_log_request.v0", ]) + def test_cultnet_replication_helpers_apply_raw_snapshot_and_shard_log(self) -> None: + document = define_database_entry_type( + "replica.item", + [ + ("name", 0), + ("category", 1), + ("value", 2, 0), + ], + cls=Item, + ) + cache = CultCache() + cache.register_document_type(document) + schema_id = document.catalog_entry().schema_id + + snapshot = { + "schemaVersion": "cultnet.snapshot_response_raw.v0", + "messageId": "snapshot-1", + "documents": [ + { + "schemaId": schema_id, + "recordKey": "item:1", + "storedAt": "2026-06-13T00:00:00Z", + "payloadEncoding": "messagepack", + "payload": document.encode_payload(Item("sword", "gear", 3)), + } + ], + } + applied_snapshot = apply_raw_snapshot(cache, [document], snapshot) + self.assertEqual(applied_snapshot[0].record_key, "item:1") + self.assertEqual(cache.get_required(document, "item:1").value, 3) + + shard_log = { + "schemaVersion": "cultnet.shard_log_response.v0", + "messageId": "log-1", + "shardId": "interop", + "shardEpoch": 1, + "resyncRequired": False, + "entries": [ + { + "sequence": 1, + "changeKind": "updated", + "put": { + "schemaVersion": "cultnet.document_put_raw.v0", + "messageId": "put-1", + "document": { + "schemaId": schema_id, + "recordKey": "item:1", + "storedAt": "2026-06-13T00:00:01Z", + "payloadEncoding": "messagepack", + "payload": document.encode_payload(Item("sword", "gear", 12)), + }, + "shardId": "interop", + "shardEpoch": 1, + }, + }, + { + "sequence": 2, + "changeKind": "removed", + "delete": { + "schemaVersion": "cultnet.document_delete.v0", + "messageId": "delete-1", + "schemaId": schema_id, + "recordKey": "item:1", + "shardId": "interop", + "shardEpoch": 1, + }, + }, + ], + } + applied_log = apply_shard_log_response(cache, [document], shard_log) + self.assertEqual([change.change_kind for change in applied_log], ["updated", "removed"]) + self.assertIsNone(cache.get(document, "item:1")) + def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None: message = document_delete( message_id="delete-1",