perf: lightweight permalink resolution to avoid eager loading

Add optimized repository methods for resolve_permalink() that skip
eager loading of observations and relations:

- permalink_exists(): Check existence without loading entity
- get_file_path_for_permalink(): Get only file_path column
- get_permalink_for_file_path(): Get only permalink column
- get_all_permalinks(): Get all permalinks as strings
- get_permalink_to_file_path_map(): Bulk lookup mapping
- get_file_path_to_permalink_map(): Reverse mapping

Updated entity_service.resolve_permalink() to use these lightweight
methods instead of loading full entities with all relationships.

Also added logfire instrumentation to markdown utils.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

src/basic_memory/markdown/entity_parser.py

@@ -22,10 +22,12 @@ from basic_memory.markdown.schemas import (
     Relation,
 )
 from basic_memory.utils import parse_tags
+import logfire
 
 md = MarkdownIt().use(observation_plugin).use(relation_plugin)
 
 
+@logfire.instrument()
 def normalize_frontmatter_value(value: Any) -> Any:
     """Normalize frontmatter values to safe types for processing.
 
@@ -87,6 +89,7 @@ def normalize_frontmatter_value(value: Any) -> Any:
     return value
 
 
+@logfire.instrument()
 def normalize_frontmatter_metadata(metadata: dict) -> dict:
     """Normalize all values in frontmatter metadata dict.
 
@@ -109,6 +112,7 @@ class EntityContent:
     relations: list[Relation] = field(default_factory=list)
 
 
+@logfire.instrument()
 def parse(content: str) -> EntityContent:
     """Parse markdown content into EntityMarkdown."""
 
@@ -167,6 +171,7 @@ class EntityParser:
                 return parsed
         return None
 
+    @logfire.instrument()
     async def parse_file(self, path: Path | str) -> EntityMarkdown:
         """Parse markdown file into EntityMarkdown."""
 
@@ -188,6 +193,7 @@ class EntityParser:
         """Get absolute path for a file using the base path for the project."""
         return self.base_path / path
 
+    @logfire.instrument()
     async def parse_file_content(self, absolute_path, file_content):
         """Parse markdown content from file stats.
 
@@ -205,6 +211,7 @@ class EntityParser:
             ctime=file_stats.st_ctime,
         )
 
+    @logfire.instrument()
     async def parse_markdown_content(
         self,
         file_path: Path,

src/basic_memory/markdown/markdown_processor.py

@@ -4,6 +4,7 @@ from collections import OrderedDict
 
 from frontmatter import Post
 from loguru import logger
+import logfire
 
 from basic_memory import file_utils
 from basic_memory.file_utils import dump_frontmatter
@@ -39,6 +40,7 @@ class MarkdownProcessor:
         """Initialize processor with base path and parser."""
         self.entity_parser = entity_parser
 
+    @logfire.instrument()
     async def read_file(self, path: Path) -> EntityMarkdown:
         """Read and parse file into EntityMarkdown schema.
 
@@ -47,6 +49,7 @@ class MarkdownProcessor:
         """
         return await self.entity_parser.parse_file(path)
 
+    @logfire.instrument()
     async def write_file(
         self,
         path: Path,
@@ -124,6 +127,7 @@ class MarkdownProcessor:
         await file_utils.write_file_atomic(path, final_content)
         return await file_utils.compute_checksum(final_content)
 
+    @logfire.instrument()
     def format_observations(self, observations: list[Observation]) -> str:
         """Format observations section in standard way.
 
@@ -132,6 +136,7 @@ class MarkdownProcessor:
         lines = [f"{obs}" for obs in observations]
         return "\n".join(lines) + "\n"
 
+    @logfire.instrument()
     def format_relations(self, relations: list[Relation]) -> str:
         """Format relations section in standard way.
 

src/basic_memory/markdown/plugins.py

@@ -32,7 +32,7 @@ def is_observation(token: Token) -> bool:
     match = re.match(r"^\[([^\[\]()]+)\]\s+(.+)", content)
     # Check for standalone hashtags (words starting with #)
     # This excludes # in HTML attributes like color="#4285F4"
-    has_tags = any(part.startswith('#') for part in content.split())
+    has_tags = any(part.startswith("#") for part in content.split())
     return bool(match) or has_tags
 
 

src/basic_memory/markdown/utils.py

@@ -2,6 +2,7 @@
 
 from pathlib import Path
 from typing import Any, Optional
+import logfire
 
 from frontmatter import Post
 
@@ -11,6 +12,7 @@ from basic_memory.models import Entity
 from basic_memory.models import Observation as ObservationModel
 
 
+@logfire.instrument()
 def entity_model_from_markdown(
     file_path: Path, markdown: EntityMarkdown, entity: Optional[Entity] = None
 ) -> Entity:
@@ -64,6 +66,7 @@ def entity_model_from_markdown(
     return model
 
 
+@logfire.instrument()
 async def schema_to_markdown(schema: Any) -> Post:
     """
     Convert schema to markdown Post object.

src/basic_memory/repository/entity_repository.py

@@ -81,6 +81,105 @@ class EntityRepository(Repository[Entity]):
         )
         return await self.find_one(query)
 
+    # -------------------------------------------------------------------------
+    # Lightweight methods for permalink resolution (no eager loading)
+    # -------------------------------------------------------------------------
+
+    @logfire.instrument()
+    async def permalink_exists(self, permalink: str) -> bool:
+        """Check if a permalink exists without loading the full entity.
+
+        This is much faster than get_by_permalink() as it skips eager loading
+        of observations and relations. Use for existence checks in bulk operations.
+
+        Args:
+            permalink: Permalink to check
+
+        Returns:
+            True if permalink exists, False otherwise
+        """
+        query = select(Entity.id).where(Entity.permalink == permalink).limit(1)
+        query = self._add_project_filter(query)
+        result = await self.execute_query(query, use_query_options=False)
+        return result.scalar_one_or_none() is not None
+
+    @logfire.instrument()
+    async def get_file_path_for_permalink(self, permalink: str) -> Optional[str]:
+        """Get the file_path for a permalink without loading the full entity.
+
+        Use when you only need the file_path, not the full entity with relations.
+
+        Args:
+            permalink: Permalink to look up
+
+        Returns:
+            file_path string if found, None otherwise
+        """
+        query = select(Entity.file_path).where(Entity.permalink == permalink)
+        query = self._add_project_filter(query)
+        result = await self.execute_query(query, use_query_options=False)
+        return result.scalar_one_or_none()
+
+    @logfire.instrument()
+    async def get_permalink_for_file_path(self, file_path: Union[Path, str]) -> Optional[str]:
+        """Get the permalink for a file_path without loading the full entity.
+
+        Use when you only need the permalink, not the full entity with relations.
+
+        Args:
+            file_path: File path to look up
+
+        Returns:
+            permalink string if found, None otherwise
+        """
+        query = select(Entity.permalink).where(Entity.file_path == Path(file_path).as_posix())
+        query = self._add_project_filter(query)
+        result = await self.execute_query(query, use_query_options=False)
+        return result.scalar_one_or_none()
+
+    @logfire.instrument()
+    async def get_all_permalinks(self) -> List[str]:
+        """Get all permalinks for this project.
+
+        Optimized for bulk operations - returns only permalink strings
+        without loading entities or relationships.
+
+        Returns:
+            List of all permalinks in the project
+        """
+        query = select(Entity.permalink)
+        query = self._add_project_filter(query)
+        result = await self.execute_query(query, use_query_options=False)
+        return list(result.scalars().all())
+
+    @logfire.instrument()
+    async def get_permalink_to_file_path_map(self) -> dict[str, str]:
+        """Get a mapping of permalink -> file_path for all entities.
+
+        Optimized for bulk permalink resolution - loads minimal data in one query.
+
+        Returns:
+            Dict mapping permalink to file_path
+        """
+        query = select(Entity.permalink, Entity.file_path)
+        query = self._add_project_filter(query)
+        result = await self.execute_query(query, use_query_options=False)
+        return {row.permalink: row.file_path for row in result.all()}
+
+    @logfire.instrument()
+    async def get_file_path_to_permalink_map(self) -> dict[str, str]:
+        """Get a mapping of file_path -> permalink for all entities.
+
+        Optimized for bulk permalink resolution - loads minimal data in one query.
+
+        Returns:
+            Dict mapping file_path to permalink
+        """
+        query = select(Entity.file_path, Entity.permalink)
+        query = self._add_project_filter(query)
+        result = await self.execute_query(query, use_query_options=False)
+        return {row.file_path: row.permalink for row in result.all()}
+
     @logfire.instrument()
     async def get_by_file_paths(
         self, session: AsyncSession, file_paths: Sequence[Union[Path, str]]

src/basic_memory/services/entity_service.py

@@ -109,6 +109,9 @@ class EntityService(BaseService[EntityModel]):
         4. Generate new unique permalink from file path
 
         Enhanced to detect and handle character-related conflicts.
+
+        Note: Uses lightweight repository methods that skip eager loading of
+        observations and relations for better performance during bulk operations.
         """
         file_path_str = Path(file_path).as_posix()
 
@@ -125,16 +128,20 @@ class EntityService(BaseService[EntityModel]):
         # If markdown has explicit permalink, try to validate it
         if markdown and markdown.frontmatter.permalink:
             desired_permalink = markdown.frontmatter.permalink
-            existing = await self.repository.get_by_permalink(desired_permalink)
+            # Use lightweight method - we only need to check file_path
+            existing_file_path = await self.repository.get_file_path_for_permalink(
+                desired_permalink
+            )
 
             # If no conflict or it's our own file, use as is
-            if not existing or existing.file_path == file_path_str:
+            if not existing_file_path or existing_file_path == file_path_str:
                 return desired_permalink
 
         # For existing files, try to find current permalink
-        existing = await self.repository.get_by_file_path(file_path_str)
-        if existing:
-            return existing.permalink
+        # Use lightweight method - we only need the permalink
+        existing_permalink = await self.repository.get_permalink_for_file_path(file_path_str)
+        if existing_permalink:
+            return existing_permalink
 
         # New file - generate permalink
         if markdown and markdown.frontmatter.permalink:
@@ -143,9 +150,10 @@ class EntityService(BaseService[EntityModel]):
             desired_permalink = generate_permalink(file_path_str)
 
         # Make unique if needed - enhanced to handle character conflicts
+        # Use lightweight existence check instead of loading full entity
         permalink = desired_permalink
         suffix = 1
-        while await self.repository.get_by_permalink(permalink):
+        while await self.repository.permalink_exists(permalink):
             permalink = f"{desired_permalink}-{suffix}"
             suffix += 1
             logger.debug(f"creating unique permalink: {permalink}")