file sync WIP

src/basic_memory/markdown/entity_parser.py

@@ -88,10 +88,10 @@ class EntityParser:
                 return parsed
         return None
 
-    async def parse_file(self, file_path: Path) -> EntityMarkdown:
+    async def parse_file(self, path: Path | str) -> EntityMarkdown:
         """Parse markdown file into EntityMarkdown."""
 
-        absolute_path = self.base_path / file_path
+        absolute_path = self.base_path / path
         # Parse frontmatter and content using python-frontmatter
         post = frontmatter.load(str(absolute_path))
 
@@ -99,7 +99,7 @@ class EntityParser:
         file_stats = absolute_path.stat()
 
         metadata = post.metadata
-        metadata["title"] = post.metadata.get("title", file_path.name)
+        metadata["title"] = post.metadata.get("title", absolute_path.name)
         metadata["type"] = post.metadata.get("type", "note")
         metadata["tags"] = parse_tags(post.metadata.get("tags", []))
 

src/basic_memory/models/knowledge.py

@@ -79,6 +79,11 @@ class Entity(Base):
         """Get all relations (incoming and outgoing) for this entity."""
         return self.incoming_relations + self.outgoing_relations
 
+    @property
+    def is_markdown(self):
+        """Check if the entity is a markdown file."""
+        return self.content_type == "text/markdown"
+
     def __repr__(self) -> str:
         return f"Entity(id={self.id}, name='{self.title}', type='{self.entity_type}'"
 

src/basic_memory/repository/search_repository.py

@@ -240,10 +240,10 @@ class SearchRepository:
         """Index or update a single item."""
         async with db.scoped_session(self.session_maker) as session:
             # Delete existing record if any
-            await session.execute(
-                text("DELETE FROM search_index WHERE permalink = :permalink"),
-                {"permalink": search_index_row.permalink},
-            )
+            # await session.execute(
+            #     text("DELETE FROM search_index WHERE permalink = :permalink"),
+            #     {"permalink": search_index_row.permalink},
+            # )
 
             # Insert new record
             await session.execute(
@@ -265,6 +265,15 @@ class SearchRepository:
             logger.debug(f"indexed row {search_index_row}")
             await session.commit()
 
+    async def delete_by_entity_id(self, entity_id: int):
+        """Delete an item from the search index by entity_id."""
+        async with db.scoped_session(self.session_maker) as session:
+            await session.execute(
+                text("DELETE FROM search_index WHERE entity_id = :entity_id"),
+                {"entity_id": entity_id},
+            )
+            await session.commit()
+
     async def delete_by_permalink(self, permalink: str):
         """Delete an item from the search index."""
         async with db.scoped_session(self.session_maker) as session:

src/basic_memory/services/entity_service.py

@@ -256,13 +256,13 @@ class EntityService(BaseService[EntityModel]):
 
     async def update_entity_relations(
         self,
-        file_path: Path,
+        path: str,
         markdown: EntityMarkdown,
     ) -> EntityModel:
         """Update relations for entity"""
-        logger.debug(f"Updating relations for entity: {file_path}")
+        logger.debug(f"Updating relations for entity: {path}")
 
-        db_entity = await self.repository.get_by_file_path(str(file_path))
+        db_entity = await self.repository.get_by_file_path(path)
 
         # Clear existing relations first
         await self.relation_repository.delete_outgoing_relations_from_entity(db_entity.id)
@@ -296,4 +296,4 @@ class EntityService(BaseService[EntityModel]):
                 )
                 continue
 
-        return await self.repository.get_by_file_path(str(file_path))
+        return await self.repository.get_by_file_path(path)

src/basic_memory/services/file_service.py

@@ -1,7 +1,8 @@
 """Service for file operations with checksum tracking."""
-
+import mimetypes
+from os import stat_result
 from pathlib import Path
-from typing import Tuple, Union
+from typing import Tuple, Union, Dict, Any
 
 from loguru import logger
 
@@ -174,3 +175,53 @@ class FileService:
         path = Path(path)
         full_path = path if path.is_absolute() else self.base_path / path
         full_path.unlink(missing_ok=True)
+
+    async def update_frontmatter(self, path: Union[Path, str], updates: Dict[str, Any]) -> str:
+        """
+        Update frontmatter fields in a file while preserving all content.
+        """
+
+        path = Path(path)
+        full_path = path if path.is_absolute() else self.base_path / path
+        return await file_utils.update_frontmatter(full_path, updates)
+
+    async def compute_checksum(self, path: Union[Path, str]) -> str:
+        """
+        Compute SHA-256 checksum of content.
+        """
+        path = Path(path)
+        full_path = path if path.is_absolute() else self.base_path / path
+        await file_utils.compute_checksum(full_path.read_text())
+        
+
+    async def file_stats(self, path: Union[Path, str]) -> stat_result:
+        """
+        Return file stats for a given path.
+        :param path: 
+        :return: 
+        """
+        path = Path(path)
+        full_path = path if path.is_absolute() else self.base_path / path
+        # get file timestamps
+        return full_path.stat()
+
+    async def content_type(self, path: Union[Path, str]) -> stat_result:
+        """
+        Return content_type for a given path.
+        :param path: 
+        :return: 
+        """
+        path = Path(path)
+        full_path = path if path.is_absolute() else self.base_path / path
+        # get file timestamps
+        mime_type, _ = mimetypes.guess_type(full_path.name)
+        content_type = mime_type or "text/plain"
+        return content_type
+
+    async def is_markdown(self, path: Union[Path, str]) -> stat_result:
+        """
+        Return content_type for a given path.
+        :param path: 
+        :return: 
+        """
+        return self.content_type(path) == "text/markdown"

src/basic_memory/services/search_service.py

@@ -118,6 +118,48 @@ class SearchService:
         self,
         entity: Entity,
         background_tasks: Optional[BackgroundTasks] = None,
+    ) -> None:
+        if background_tasks:
+            background_tasks.add_task(self.index_entity_data, entity)
+        else:
+            await self.index_entity_data(entity)
+
+    async def index_entity_data(
+        self,
+        entity: Entity,
+    ) -> None:
+        
+        # delete all search index data associated with entity
+        await self.repository.delete_by_entity_id(entity_id=entity.id)
+        
+        # reindex
+        await self.index_entity_markdown(
+            entity
+        ) if entity.is_markdown else await self.index_entity_file(entity)
+
+    async def index_entity_file(
+        self,
+        entity: Entity,
+    ) -> None:
+        # Index entity file with no content
+        await self.repository.index_item(
+            SearchIndexRow(
+                id=entity.id,
+                type=SearchItemType.ENTITY.value,
+                title=entity.title,
+                permalink=entity.permalink,
+                file_path=entity.file_path,
+                metadata={
+                    "entity_type": entity.entity_type,
+                },
+                created_at=entity.created_at,
+                updated_at=entity.updated_at,
+            )
+        )
+
+    async def index_entity_markdown(
+        self,
+        entity: Entity,
     ) -> None:
         """Index an entity and all its observations and relations.
 
@@ -136,16 +178,6 @@ class SearchService:
 
         Each type gets its own row in the search index with appropriate metadata.
         """
-        if background_tasks:
-            background_tasks.add_task(self.index_entity_data, entity)
-        else:
-            await self.index_entity_data(entity)
-
-    async def index_entity_data(
-        self,
-        entity: Entity,
-    ) -> None:
-        """Actually perform the indexing."""
 
         content_parts = []
         title_variants = self._generate_variants(entity.title)
@@ -169,6 +201,7 @@ class SearchService:
                 content=entity_content,
                 permalink=entity.permalink,
                 file_path=entity.file_path,
+                entity_id=entity.id,
                 metadata={
                     "entity_type": entity.entity_type,
                 },
@@ -214,6 +247,7 @@ class SearchService:
                     permalink=rel.permalink,
                     file_path=entity.file_path,
                     type=SearchItemType.RELATION.value,
+                    entity_id=entity.id,
                     from_id=rel.from_id,
                     to_id=rel.to_id,
                     relation_type=rel.relation_type,

src/basic_memory/sync/sync_service.py

@@ -1,28 +1,25 @@
 """Service for syncing files between filesystem and database."""
 
+import mimetypes
 from pathlib import Path
-from typing import Dict
+from typing import Tuple
 
 import logfire
 from loguru import logger
 from sqlalchemy.exc import IntegrityError
 
 from basic_memory import file_utils
-from basic_memory.markdown import EntityParser, EntityMarkdown
+from basic_memory.markdown import EntityParser
 from basic_memory.repository import EntityRepository, RelationRepository
-from basic_memory.services import EntityService
+from basic_memory.services import EntityService, FileService
 from basic_memory.services.search_service import SearchService
 from basic_memory.sync import FileChangeScanner
 from basic_memory.sync.utils import SyncReport
+from basic_memory.models import Entity
 
 
 class SyncService:
-    """Syncs documents and knowledge files with database.
-
-    Implements two-pass sync strategy for knowledge files to handle relations:
-    1. First pass creates/updates entities without relations
-    2. Second pass processes relations after all entities exist
-    """
+    """Syncs documents and knowledge files with database."""
 
     def __init__(
         self,
@@ -32,6 +29,7 @@ class SyncService:
         entity_repository: EntityRepository,
         relation_repository: RelationRepository,
         search_service: SearchService,
+        file_service: FileService,
     ):
         self.scanner = scanner
         self.entity_service = entity_service
@@ -39,9 +37,87 @@ class SyncService:
         self.entity_repository = entity_repository
         self.relation_repository = relation_repository
         self.search_service = search_service
+        self.file_service = file_service
+
+    async def sync_file(self, path: str) -> Tuple[Entity, str]:
+        """Sync a single file completely."""
+
+        try:
+            if self.file_service.is_markdown(path):
+                entity, checksum = await self.sync_markdown_file(path)
+            else:
+                entity, checksum = await self.sync_regular_file(path)
+            await self.search_service.index_entity(entity)
+            return entity, checksum
+
+        except Exception as e:
+            logger.error(f"Failed to sync {path}: {e}")
+            raise
+
+    async def sync_markdown_file(self, path: str) -> Tuple[Entity, str]:
+        """Sync a markdown file with full processing."""
+
+        # Parse markdown first to get any existing permalink
+        entity_markdown = await self.entity_parser.parse_file(path)
+
+        # Resolve permalink - this handles all the cases including conflicts
+        permalink = await self.entity_service.resolve_permalink(path, markdown=entity_markdown)
+
+        # If permalink changed, update the file
+        if permalink != entity_markdown.frontmatter.permalink:
+            logger.info(f"Updating permalink in {path}: {permalink}")
+            entity_markdown.frontmatter.metadata["permalink"] = permalink
+            checksum = await self.file_service.update_frontmatter(path, {"permalink": permalink})
+        else:
+            checksum = await self.file_service.compute_checksum(path)
+
+        # Create/update entity with resolved permalink
+        entity = await self.entity_service.create_entity_from_markdown(path, entity_markdown)
+
+        # Update relations and search index
+        entity = await self.entity_service.update_entity_relations(path, entity_markdown)
+
+        return entity, checksum
+
+    async def sync_regular_file(self, path: Path) -> Tuple[Entity, str]:
+        """Sync a non-markdown file with basic tracking."""
+
+        checksum = await self.file_service.compute_checksum(path)
+        existing = await self.entity_repository.get_by_file_path(path)
+        if not existing:
+            # Generate permalink from path
+            permalink = await self.entity_service.resolve_permalink(path)
+
+            # get file timestamps
+            file_stats = self.file_service.file_stats(path)
+
+            # get mime type
+            mime_type, _ = mimetypes.guess_type(path.name)
+            content_type = mime_type or "text/plain"
+
+            entity = await self.entity_repository.add(
+                Entity(
+                    entity_type="file",
+                    file_path=path,
+                    permalink=permalink,
+                    checksum=checksum,
+                    title=path.name,
+                    created_at=file_stats.st_ctime,
+                    updated_at=file_stats.st_mtime,
+                    content_type=content_type,
+                )
+            )
+        else:
+            entity = await self.entity_repository.update(
+                existing.id, {"file_path": path, "checksum": checksum}
+            )
+
+        await self.search_service.index_entity(entity)
+        return entity, checksum
 
     async def handle_entity_deletion(self, file_path: str):
         """Handle complete entity deletion including search index cleanup."""
+
         # First get entity to get permalink before deletion
         entity = await self.entity_repository.get_by_file_path(file_path)
         if entity:
@@ -61,9 +137,9 @@ class SyncService:
                 await self.search_service.delete_by_permalink(permalink)
 
     async def sync(self, directory: Path) -> SyncReport:
-        """Sync knowledge files with database."""
+        """Sync all files with database."""
 
-        with logfire.span("sync", directory=directory):  # pyright: ignore [reportGeneralTypeIssues]
+        with logfire.span("sync", directory=directory):
             changes = await self.scanner.find_knowledge_changes(directory)
             logger.info(f"Found {changes.total_changes} knowledge changes")
 
@@ -73,102 +149,47 @@ class SyncService:
                 entity = await self.entity_repository.get_by_file_path(old_path)
                 if entity:
                     # Update file_path but keep the same permalink for link stability
-                    updated = await self.entity_repository.update(
+                    await self.entity_repository.update(
                         entity.id, {"file_path": new_path, "checksum": changes.checksums[new_path]}
                     )
                     # update search index
-                    if updated:
-                        await self.search_service.index_entity(updated)
+                    await self.search_service.index_entity(entity)
 
             # Handle deletions next
-            # remove rows from db for files no longer present
             for path in changes.deleted:
                 await self.handle_entity_deletion(path)
 
-            # Parse files that need updating
-            parsed_entities: Dict[str, EntityMarkdown] = {}
-
+            # Handle new and modified files
             for path in [*changes.new, *changes.modified]:
-                entity_markdown = await self.entity_parser.parse_file(directory / path)
-                parsed_entities[path] = entity_markdown
-
-            # First pass: Create/update entities
-            # entities will have a null checksum to indicate they are not complete
-            for path, entity_markdown in parsed_entities.items():
-                # Get unique permalink and update markdown if needed
-                permalink = await self.entity_service.resolve_permalink(
-                    Path(path), markdown=entity_markdown
-                )
-
-                if permalink != entity_markdown.frontmatter.permalink:
-                    # Add/update permalink in frontmatter
-                    logger.info(f"Adding permalink '{permalink}' to file: {path}")
-
-                    # update markdown
-                    entity_markdown.frontmatter.metadata["permalink"] = permalink
-
-                    # update file frontmatter
-                    updated_checksum = await file_utils.update_frontmatter(
-                        directory / path, {"permalink": permalink}
-                    )
-
-                    # Update checksum in changes report since file was modified
-                    changes.checksums[path] = updated_checksum
-
-                # if the file is new, create an entity
-                if path in changes.new:
-                    # Create entity with final permalink
-                    logger.debug(f"Creating new entity_markdown: {path}")
-                    await self.entity_service.create_entity_from_markdown(
-                        Path(path), entity_markdown
-                    )
-                # otherwise we need to update the entity and observations
-                else:
-                    logger.debug(f"Updating entity_markdown: {path}")
-                    await self.entity_service.update_entity_and_observations(
-                        Path(path), entity_markdown
-                    )
-
-            # Second pass
-            for path, entity_markdown in parsed_entities.items():
-                logger.debug(f"Updating relations for: {path}")
-
-                # Process relations
-                checksum = changes.checksums[path]
-                entity = await self.entity_service.update_entity_relations(
-                    Path(path), entity_markdown
-                )
-
-                # add to search index
-                await self.search_service.index_entity(entity)
-
-                # Set final checksum to mark sync complete
-                await self.entity_repository.update(entity.id, {"checksum": checksum})
-
-            # Third pass: Try to resolve any forward references
-            logger.debug("Attempting to resolve forward references")
-            for relation in await self.relation_repository.find_unresolved_relations():
-                target_entity = await self.entity_service.link_resolver.resolve_link(
-                    relation.to_name
-                )
-                # check we found a link that is not the source
-                if target_entity and target_entity.id != relation.from_id:
-                    logger.debug(
-                        f"Resolved forward reference: {relation.to_name} -> {target_entity.permalink}"
-                    )
-
-                    try:
-                        await self.relation_repository.update(
-                            relation.id,
-                            {
-                                "to_id": target_entity.id,
-                                "to_name": target_entity.title,  # Update to actual title
-                            },
-                        )
-                    except IntegrityError:
-                        logger.debug(f"Ignoring duplicate relation {relation}")
-
-                    # update search index
-                    await self.search_service.index_entity(target_entity)
+                logger.debug(f"Syncing file: {path}")
+                entity, checksum = await self.sync_file(path)
+                changes.checksums[path] = checksum
 
+            await self.resolve_relations()
             return changes
+
+    async def resolve_relations(self):
+        """Try to resolve any unresolved relations"""
+
+        logger.debug("Attempting to resolve forward references")
+        for relation in await self.relation_repository.find_unresolved_relations():
+            resolved_entity = await self.entity_service.link_resolver.resolve_link(relation.to_name)
+
+            # ignore reference to self
+            if resolved_entity and resolved_entity.id != relation.from_id:
+                logger.debug(
+                    f"Resolved forward reference: {relation.to_name} -> {resolved_entity.title}"
+                )
+                try:
+                    await self.relation_repository.update(
+                        relation.id,
+                        {
+                            "to_id": resolved_entity.id,
+                            "to_name": resolved_entity.title,
+                        },
+                    )
+                except IntegrityError:
+                    logger.debug(f"Ignoring duplicate relation {relation}")
+
+                # update search index
+                await self.search_service.index_entity(resolved_entity)

src/basic_memory/sync/sync_service.py.bak

@@ -0,0 +1,174 @@
+"""Service for syncing files between filesystem and database."""
+
+from pathlib import Path
+from typing import Dict
+
+import logfire
+from loguru import logger
+from sqlalchemy.exc import IntegrityError
+
+from basic_memory import file_utils
+from basic_memory.markdown import EntityParser, EntityMarkdown
+from basic_memory.repository import EntityRepository, RelationRepository
+from basic_memory.services import EntityService
+from basic_memory.services.search_service import SearchService
+from basic_memory.sync import FileChangeScanner
+from basic_memory.sync.utils import SyncReport
+
+
+class SyncService:
+    """Syncs documents and knowledge files with database.
+
+    Implements two-pass sync strategy for knowledge files to handle relations:
+    1. First pass creates/updates entities without relations
+    2. Second pass processes relations after all entities exist
+    """
+
+    def __init__(
+        self,
+        scanner: FileChangeScanner,
+        entity_service: EntityService,
+        entity_parser: EntityParser,
+        entity_repository: EntityRepository,
+        relation_repository: RelationRepository,
+        search_service: SearchService,
+    ):
+        self.scanner = scanner
+        self.entity_service = entity_service
+        self.entity_parser = entity_parser
+        self.entity_repository = entity_repository
+        self.relation_repository = relation_repository
+        self.search_service = search_service
+
+    async def handle_entity_deletion(self, file_path: str):
+        """Handle complete entity deletion including search index cleanup."""
+        # First get entity to get permalink before deletion
+        entity = await self.entity_repository.get_by_file_path(file_path)
+        if entity:
+            logger.debug(f"Deleting entity and cleaning up search index: {file_path}")
+
+            # Delete from db (this cascades to observations/relations)
+            await self.entity_service.delete_entity_by_file_path(file_path)
+
+            # Clean up search index
+            permalinks = (
+                [entity.permalink]
+                + [o.permalink for o in entity.observations]
+                + [r.permalink for r in entity.relations]
+            )
+            logger.debug(f"Deleting from search index: {permalinks}")
+            for permalink in permalinks:
+                await self.search_service.delete_by_permalink(permalink)
+
+    async def sync(self, directory: Path) -> SyncReport:
+        """Sync knowledge files with database."""
+
+        with logfire.span("sync", directory=directory):  # pyright: ignore [reportGeneralTypeIssues]
+            changes = await self.scanner.find_knowledge_changes(directory)
+            logger.info(f"Found {changes.total_changes} knowledge changes")
+
+            # Handle moves first
+            for old_path, new_path in changes.moves.items():
+                logger.debug(f"Moving entity: {old_path} -> {new_path}")
+                entity = await self.entity_repository.get_by_file_path(old_path)
+                if entity:
+                    # Update file_path but keep the same permalink for link stability
+                    updated = await self.entity_repository.update(
+                        entity.id, {"file_path": new_path, "checksum": changes.checksums[new_path]}
+                    )
+                    # update search index
+                    if updated:
+                        await self.search_service.index_entity(updated)
+
+            # Handle deletions next
+            # remove rows from db for files no longer present
+            for path in changes.deleted:
+                await self.handle_entity_deletion(path)
+
+            # Parse files that need updating
+            parsed_entities: Dict[str, EntityMarkdown] = {}
+
+            for path in [*changes.new, *changes.modified]:
+                entity_markdown = await self.entity_parser.parse_file(directory / path)
+                parsed_entities[path] = entity_markdown
+
+            # First pass: Create/update entities
+            # entities will have a null checksum to indicate they are not complete
+            for path, entity_markdown in parsed_entities.items():
+                # Get unique permalink and update markdown if needed
+                permalink = await self.entity_service.resolve_permalink(
+                    Path(path), markdown=entity_markdown
+                )
+
+                if permalink != entity_markdown.frontmatter.permalink:
+                    # Add/update permalink in frontmatter
+                    logger.info(f"Adding permalink '{permalink}' to file: {path}")
+
+                    # update markdown
+                    entity_markdown.frontmatter.metadata["permalink"] = permalink
+
+                    # update file frontmatter
+                    updated_checksum = await file_utils.update_frontmatter(
+                        directory / path, {"permalink": permalink}
+                    )
+
+                    # Update checksum in changes report since file was modified
+                    changes.checksums[path] = updated_checksum
+
+                # if the file is new, create an entity
+                if path in changes.new:
+                    # Create entity with final permalink
+                    logger.debug(f"Creating new entity_markdown: {path}")
+                    await self.entity_service.create_entity_from_markdown(
+                        Path(path), entity_markdown
+                    )
+                # otherwise we need to update the entity and observations
+                else:
+                    logger.debug(f"Updating entity_markdown: {path}")
+                    await self.entity_service.update_entity_and_observations(
+                        Path(path), entity_markdown
+                    )
+
+            # Second pass
+            for path, entity_markdown in parsed_entities.items():
+                logger.debug(f"Updating relations for: {path}")
+
+                # Process relations
+                checksum = changes.checksums[path]
+                entity = await self.entity_service.update_entity_relations(
+                    Path(path), entity_markdown
+                )
+
+                # add to search index
+                await self.search_service.index_entity(entity)
+
+                # Set final checksum to mark sync complete
+                await self.entity_repository.update(entity.id, {"checksum": checksum})
+
+            # Third pass: Try to resolve any forward references
+            logger.debug("Attempting to resolve forward references")
+            for relation in await self.relation_repository.find_unresolved_relations():
+                target_entity = await self.entity_service.link_resolver.resolve_link(
+                    relation.to_name
+                )
+                # check we found a link that is not the source
+                if target_entity and target_entity.id != relation.from_id:
+                    logger.debug(
+                        f"Resolved forward reference: {relation.to_name} -> {target_entity.permalink}"
+                    )
+
+                    try:
+                        await self.relation_repository.update(
+                            relation.id,
+                            {
+                                "to_id": target_entity.id,
+                                "to_name": target_entity.title,  # Update to actual title
+                            },
+                        )
+                    except IntegrityError:
+                        logger.debug(f"Ignoring duplicate relation {relation}")
+
+                    # update search index
+                    await self.search_service.index_entity(target_entity)
+
+            return changes