diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index cb5b2cf7..e9150d2d 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -535,6 +535,8 @@ class CultMeshDatabase: ) -> CultNetRawSnapshotResponse: requested_schema_ids = set(schema_ids or []) requested_record_keys = set(record_keys or []) + documents_by_type = {document.type: document for document in self.documents} + documents_by_schema_id = schema_document_map(self.documents) shard_record_keys = ( self._live_shard_record_keys(shard_id) if shard_id is not None and shard_id in self._shard_logs @@ -543,7 +545,8 @@ class CultMeshDatabase: documents: list[CultNetRawDocumentRecord] = [] for envelope in self.cache.snapshot_envelopes(): schema_id = envelope.schema_id or envelope.type - if requested_schema_ids and schema_id not in requested_schema_ids: + document = documents_by_type.get(envelope.type) or documents_by_schema_id.get(schema_id) + if requested_schema_ids and not self._schema_matches_request(schema_id, document, requested_schema_ids): continue if requested_record_keys and envelope.key not in requested_record_keys: continue @@ -950,6 +953,31 @@ class CultMeshDatabase: previous[(resolved_schema_id, record_key)] = self.cache.get(document, record_key) return previous + @staticmethod + def _schema_matches_request( + schema_id: str, + document: DocumentDefinition[Any] | None, + requested_schema_ids: set[str], + ) -> bool: + if not requested_schema_ids or schema_id in requested_schema_ids: + return True + if document is None: + return False + entry = document.catalog_entry() + if entry.schema_id in requested_schema_ids or entry.schema_name in requested_schema_ids: + return True + if any(schema_id in requested_schema_ids for schema_id in entry.compatible_schema_ids): + return True + return any(_infer_schema_name(schema_id) == entry.schema_name for schema_id in requested_schema_ids) + + +def _infer_schema_name(schema_id: str) -> str | None: + marker = schema_id.rfind(".v") + if marker <= 0 or marker + 2 >= len(schema_id): + return None + version = schema_id[marker + 2:] + return schema_id[:marker] if version.isdigit() else None + def create_node( cache_path: str | Path | None = None, diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index 986460ff..088c335d 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -591,12 +591,21 @@ class CultMeshLocalServer: def _shard_catalog_response(self, request: dict[str, Any]) -> dict[str, Any]: requested_schema_ids = {str(value) for value in request.get("schemaIds") or []} document_schema_ids = [document.catalog_entry().schema_id for document in self.node.documents] + documents_by_schema_id = _schema_document_map(self.node.documents) shard_ids = self.node.database.shard_ids() or ["primary"] shards = [] for shard_id in shard_ids: schema_ids = self.node.database.shard_schema_ids(shard_id) or document_schema_ids if requested_schema_ids: - schema_ids = [schema_id for schema_id in schema_ids if schema_id in requested_schema_ids] + schema_ids = [ + schema_id + for schema_id in schema_ids + if _schema_matches_request( + schema_id, + documents_by_schema_id.get(schema_id), + requested_schema_ids, + ) + ] if not schema_ids: continue shards.append({ @@ -620,3 +629,39 @@ class CultMeshLocalServer: def _now_ms() -> int: return int(time.time() * 1000) + + +def _schema_document_map(documents: list[Any]) -> dict[str, Any]: + mapped: dict[str, Any] = {} + for document in documents: + entry = document.catalog_entry() + mapped[entry.schema_id] = document + mapped[entry.schema_name] = document + for compatible_schema_id in entry.compatible_schema_ids: + mapped[compatible_schema_id] = document + return mapped + + +def _schema_matches_request( + schema_id: str, + document: Any | None, + requested_schema_ids: set[str], +) -> bool: + if not requested_schema_ids or schema_id in requested_schema_ids: + return True + if document is None: + return False + entry = document.catalog_entry() + if entry.schema_id in requested_schema_ids or entry.schema_name in requested_schema_ids: + return True + if any(compatible_schema_id in requested_schema_ids for compatible_schema_id in entry.compatible_schema_ids): + return True + return any(_infer_schema_name(requested_schema_id) == entry.schema_name for requested_schema_id in requested_schema_ids) + + +def _infer_schema_name(schema_id: str) -> str | None: + marker = schema_id.rfind(".v") + if marker <= 0 or marker + 2 >= len(schema_id): + return None + version = schema_id[marker + 2:] + return schema_id[:marker] if version.isdigit() else None diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index c45bafad..cc8c85ac 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -5233,6 +5233,40 @@ class CultCacheTests(unittest.TestCase): self.assertIsNone(subscription_delete.raw_document) self.assertIsNone(node.database.get(document, "note:2")) + def test_cultmesh_local_server_resolves_snapshot_and_shard_catalog_schema_aliases(self) -> None: + document = define_database_entry_type( + "mesh.alias_server_note", + [("body", 0)], + schema_id="sha256:mesh-alias-server-note", + schema_name="mesh.alias_server_note", + schema_version="mesh.alias_server_note.v1", + ) + node = CultMesh.create_node(runtime_id="mesh-alias-server") + node.database.register_document(document) + node.database.put_raw_message( + document, + "note:alias", + {"body": "served through alias"}, + shard_id="notes", + shard_epoch=1, + ) + + server = CultMesh.serve_node(node) + try: + raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0) + snapshot = raw_client.fetch_snapshot(schema_ids=["mesh.alias_server_note.v1"]) + typed_snapshot = raw_client.fetch_snapshot_response(schema_ids=["mesh.alias_server_note.v1"]) + shard_catalog = raw_client.fetch_shard_catalog(schema_ids=["mesh.alias_server_note.v1"]) + finally: + server.stop() + + local_schema_id = document.catalog_entry().schema_id + self.assertEqual(snapshot["documents"][0]["schemaId"], local_schema_id) + self.assertEqual(snapshot["documents"][0]["recordKey"], "note:alias") + self.assertEqual(typed_snapshot.documents[0].schema_id, local_schema_id) + self.assertEqual(shard_catalog["shards"][0]["schemaIds"], [local_schema_id]) + self.assertEqual(shard_catalog["shards"][0]["shardId"], "notes") + def test_cultmesh_local_server_rejects_oversized_snapshot_responses(self) -> None: document = define_database_entry_type( "mesh.snapshot_limit_note",