Speed up stored snapshot inventory

packages/storage-core/src/storage_core/__init__.py

@@ -45,12 +45,35 @@ class FileStorage:
         self.initialize()
         result: list[StoredSnapshotInfo] = []
         for path in sorted(self.snapshots_dir.glob("*.json")):
-            snapshot = snapshot_from_json(path.read_bytes())
+            try:
+                payload = orjson.loads(path.read_bytes())
+            except (FileNotFoundError, orjson.JSONDecodeError):
+                continue
+            project_id = payload.get("project_id")
+            snapshot_id = payload.get("snapshot_id")
+            if not isinstance(project_id, str) or not isinstance(snapshot_id, str):
+                continue
+            snapshot_hash = payload.get("snapshot_hash")
             result.append(
                 StoredSnapshotInfo(
-                    project_id=snapshot.project_id,
-                    snapshot_id=snapshot.snapshot_id,
-                    snapshot_hash=snapshot.snapshot_hash,
+                    project_id=project_id,
+                    snapshot_id=snapshot_id,
+                    snapshot_hash=snapshot_hash if isinstance(snapshot_hash, str) else None,
+                    path=path.as_posix(),
+                )
+            )
+        return result
+
+    def list_snapshot_refs(self) -> list[StoredSnapshotInfo]:
+        self.initialize()
+        result: list[StoredSnapshotInfo] = []
+        for path in sorted(self.snapshots_dir.glob("*.json")):
+            project_id = path.stem
+            result.append(
+                StoredSnapshotInfo(
+                    project_id=project_id,
+                    snapshot_id=f"stored.{project_id}",
+                    snapshot_hash=None,
                     path=path.as_posix(),
                 )
             )
@@ -59,6 +82,11 @@ class FileStorage:
     def has_snapshot(self, project_id: str) -> bool:
         return self._snapshot_path(project_id).exists()
 
+    def count_snapshots(self) -> int:
+        if not self.snapshots_dir.exists():
+            return 0
+        return sum(1 for _ in self.snapshots_dir.glob("*.json"))
+
     def write_document(self, collection: str, document_id: str, payload: dict[str, Any]) -> Path:
         directory = self._collection_dir(collection)
         directory.mkdir(parents=True, exist_ok=True)

packages/storage-core/tests/test_storage_core.py

@@ -17,6 +17,10 @@ def test_file_storage_saves_and_loads_snapshot(tmp_path: Path):
     assert info.project_id == "demo"
     assert restored.snapshot_hash == snapshot.snapshot_hash
     assert storage.list_snapshots()[0].snapshot_id == snapshot.snapshot_id
+    snapshot_ref = storage.list_snapshot_refs()[0]
+    assert snapshot_ref.project_id == "demo"
+    assert snapshot_ref.snapshot_id == "stored.demo"
+    assert snapshot_ref.path.endswith("demo.json")
 
 
 def test_file_storage_generic_documents(tmp_path: Path):