diff --git a/src/basic_memory/config.py b/src/basic_memory/config.py index a8d2264f..2ec108fa 100644 --- a/src/basic_memory/config.py +++ b/src/basic_memory/config.py @@ -35,7 +35,7 @@ class ProjectConfig(BaseSettings): default=500, description="Milliseconds to wait after changes before syncing", gt=0 ) - log_level: str = "INFO" + log_level: str = "DEBUG" model_config = SettingsConfigDict( env_prefix="BASIC_MEMORY_", diff --git a/src/basic_memory/models/knowledge.py b/src/basic_memory/models/knowledge.py index cf9f3cbb..0a4fa680 100644 --- a/src/basic_memory/models/knowledge.py +++ b/src/basic_memory/models/knowledge.py @@ -132,7 +132,8 @@ class Relation(Base): __tablename__ = "relation" __table_args__ = ( - UniqueConstraint("from_id", "to_id", "relation_type", name="uix_relation"), + UniqueConstraint("from_id", "to_id", "relation_type", name="uix_relation_from_id_to_id"), + UniqueConstraint("from_id", "to_name", "relation_type", name="uix_relation_from_id_to_name"), Index("ix_relation_type", "relation_type"), Index("ix_relation_from_id", "from_id"), # Add FK indexes Index("ix_relation_to_id", "to_id"), diff --git a/src/basic_memory/repository/entity_repository.py b/src/basic_memory/repository/entity_repository.py index fac438a8..f41d17e2 100644 --- a/src/basic_memory/repository/entity_repository.py +++ b/src/basic_memory/repository/entity_repository.py @@ -1,8 +1,9 @@ """Repository for managing entities in the knowledge graph.""" from pathlib import Path -from typing import List, Optional, Sequence, Union +from typing import List, Optional, Sequence, Union, Dict +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.orm import selectinload from sqlalchemy.orm.interfaces import LoaderOption diff --git a/src/basic_memory/repository/repository.py b/src/basic_memory/repository/repository.py index 285256bd..2c367765 100644 --- a/src/basic_memory/repository/repository.py +++ b/src/basic_memory/repository/repository.py @@ -97,7 +97,7 @@ class Repository[T: Base]: entities = (self.Model,) return select(*entities) - async def find_all(self, skip: int = 0, limit: Optional[int] = 0) -> Sequence[T]: + async def find_all(self, skip: int = 0, limit: int = 10) -> Sequence[T]: """Fetch records from the database with pagination.""" logger.debug(f"Finding all {self.Model.__name__} (skip={skip}, limit={limit})") diff --git a/src/basic_memory/services/file_service.py b/src/basic_memory/services/file_service.py index dd3b148a..6d439075 100644 --- a/src/basic_memory/services/file_service.py +++ b/src/basic_memory/services/file_service.py @@ -191,7 +191,9 @@ class FileService: """ path = Path(path) full_path = path if path.is_absolute() else self.base_path / path - await file_utils.compute_checksum(full_path.read_text()) + + # TODO checksum for binary files + return await file_utils.compute_checksum(full_path.read_text()) async def file_stats(self, path: Union[Path, str]) -> stat_result: diff --git a/src/basic_memory/sync/file_change_scanner.py b/src/basic_memory/sync/file_change_scanner.py index d7710aca..d6da834a 100644 --- a/src/basic_memory/sync/file_change_scanner.py +++ b/src/basic_memory/sync/file_change_scanner.py @@ -27,6 +27,10 @@ class ScanResult: # file_path -> checksum files: Dict[str, str] = field(default_factory=dict) + + # checksum -> file_path + checksums: Dict[str, str] = field(default_factory=dict) + # file_path -> error message errors: Dict[str, str] = field(default_factory=dict) @@ -155,4 +159,4 @@ class FileChangeScanner: async def find_knowledge_changes(self, directory: Path) -> SyncReport: """Find changes in knowledge directory.""" db_file_state = await self.get_db_file_state(await self.entity_repository.find_all()) - return await self.find_changes(directory=directory, db_file_state=db_file_state) + return await self.find_changes(directory=directory, db_file_state=db_file_state) \ No newline at end of file diff --git a/src/basic_memory/sync/sync_service.py b/src/basic_memory/sync/sync_service.py index 48a29cb1..2bba0d8f 100644 --- a/src/basic_memory/sync/sync_service.py +++ b/src/basic_memory/sync/sync_service.py @@ -1,21 +1,72 @@ """Service for syncing files between filesystem and database.""" import mimetypes +from dataclasses import dataclass +from dataclasses import field from pathlib import Path +from typing import Set, Dict, Sequence 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 +from basic_memory.models import Entity from basic_memory.repository import EntityRepository, RelationRepository 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 + + +@dataclass +class SyncReport: + """Report of file changes found compared to database state. + + Attributes: + total: Total number of files in directory being synced + new: Files that exist on disk but not in database + modified: Files that exist in both but have different checksums + deleted: Files that exist in database but not on disk + moves: Files that have been moved from one location to another + checksums: Current checksums for files on disk + """ + + total: int = 0 + # We keep paths as strings in sets/dicts for easier serialization + new: Set[str] = field(default_factory=set) + modified: Set[str] = field(default_factory=set) + deleted: Set[str] = field(default_factory=set) + moves: Dict[str, str] = field(default_factory=dict) # old_path -> new_path + + @property + def total_changes(self) -> int: + """Total number of changes.""" + return len(self.new) + len(self.modified) + len(self.deleted) + len(self.moves) + + +@dataclass +class ScanResult: + """Result of scanning a directory.""" + + # file_path -> checksum + files: Dict[str, str] = field(default_factory=dict) + + # checksum -> file_path + checksums: Dict[str, str] = field(default_factory=dict) + + # file_path -> error message + errors: Dict[str, str] = field(default_factory=dict) + + +@dataclass +class FileState: + """State of a file including file path, permalink and checksum info.""" + + file_path: str + permalink: str + checksum: str class SyncService: @@ -39,14 +90,97 @@ class SyncService: 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.""" + async def get_db_file_state(self) -> Dict[str, FileState]: + """Get file_path and checksums from database. + Args: + db_records: database records + Returns: + Dict mapping file paths to FileState + :param db_records: the data from the db + """ + db_records = await self.entity_repository.find_all() + return { + r.file_path: FileState( + file_path=r.file_path, permalink=r.permalink, checksum=r.checksum or "" + ) + for r in db_records + } + + async def sync(self, directory: Path) -> SyncReport: + """Sync all files with database.""" + + with logfire.span("sync", directory=directory): + # initial paths from db to sync + # path -> FileState + db_paths = await self.get_db_file_state() + + # Track potentially moved files by checksum + + scan_result = await self.scan_directory(directory) + + files_by_checksum = scan_result.checksums + + report = SyncReport() + + # First find potential new files and record checksums + # if a path is not present in the db, it could be new or could be the destination of a move + for file_path, checksum in scan_result.files.items(): + if file_path not in db_paths: + report.new.add(file_path) + + # Now detect moves and deletions + for db_path, db_state in db_paths.items(): + local_checksum_for_db_path = scan_result.files.get(db_path) + + # file not modified + if db_state.checksum == local_checksum_for_db_path: + pass + + # if checksums don't match for the same path, its modified + if local_checksum_for_db_path and db_state.checksum != local_checksum_for_db_path: + report.modified.add(db_path) + + # check if it's moved or deleted + if not local_checksum_for_db_path: + # if we find the checksum in another file, it's a move + if db_state.checksum in files_by_checksum: + new_path = files_by_checksum[db_state.checksum] + report.moves[db_path] = new_path + # Remove from new files since it's a move + report.new.remove(new_path) + + # deleted + else: + report.deleted.add(db_path) + + # order of sync matters to resolve relations effectively + + # sync moves first + for old_path, new_path in report.moves.items(): + await self.handle_move(old_path, new_path) + + # deleted next + for path in report.deleted: + await self.handle_delete(path) + + # then new and modified + for path in report.new: + await self.sync_file(path, new=True) + + for path in report.modified: + await self.sync_file(path, new=False) + + await self.resolve_relations() + return report + + async def sync_file(self, path: str, new: bool = True) -> Tuple[Entity, str]: + """Sync a single file.""" try: if self.file_service.is_markdown(path): - entity, checksum = await self.sync_markdown_file(path) + entity, checksum = await self.sync_markdown_file(path, new) else: - entity, checksum = await self.sync_regular_file(path) + entity, checksum = await self.sync_regular_file(path, new) await self.search_service.index_entity(entity) return entity, checksum @@ -54,7 +188,7 @@ class SyncService: logger.error(f"Failed to sync {path}: {e}") raise - async def sync_markdown_file(self, path: str) -> Tuple[Entity, str]: + async def sync_markdown_file(self, path: str, new: bool = True) -> Tuple[Entity, str]: """Sync a markdown file with full processing.""" # Parse markdown first to get any existing permalink @@ -71,20 +205,34 @@ class SyncService: 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) + # if the file is new, create an entity + if new: + # Create entity with final permalink + logger.debug(f"Creating new entity from 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 from markdown: {path}") + await self.entity_service.update_entity_and_observations( + Path(path), entity_markdown + ) + # Update relations and search index entity = await self.entity_service.update_entity_relations(path, entity_markdown) - + + # set checksum + await self.entity_repository.update(entity.id, {"checksum": checksum}) return entity, checksum - async def sync_regular_file(self, path: Path) -> Tuple[Entity, str]: + async def sync_regular_file(self, path: Path, new: bool = True) -> 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: + if new: # Generate permalink from path permalink = await self.entity_service.resolve_permalink(path) @@ -108,14 +256,15 @@ class SyncService: ) ) else: + entity = await self.entity_repository.get_by_file_path(path) entity = await self.entity_repository.update( - existing.id, {"file_path": path, "checksum": checksum} + entity.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): + async def handle_delete(self, file_path: str): """Handle complete entity deletion including search index cleanup.""" # First get entity to get permalink before deletion @@ -136,37 +285,16 @@ class SyncService: for permalink in permalinks: await self.search_service.delete_by_permalink(permalink) - async def sync(self, directory: Path) -> SyncReport: - """Sync all files with database.""" - - with logfire.span("sync", directory=directory): - 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 - await self.entity_repository.update( - entity.id, {"file_path": new_path, "checksum": changes.checksums[new_path]} - ) - # update search index - await self.search_service.index_entity(entity) - - # Handle deletions next - for path in changes.deleted: - await self.handle_entity_deletion(path) - - # Handle new and modified files - for path in [*changes.new, *changes.modified]: - 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 handle_move(self, old_path, new_path ): + 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} + ) + # update search index + await self.search_service.index_entity(updated) async def resolve_relations(self): """Try to resolve any unresolved relations""" @@ -193,3 +321,45 @@ class SyncService: # update search index await self.search_service.index_entity(resolved_entity) + + async def scan_directory(self, directory: Path) -> ScanResult: + """ + Scan directory for markdown files and their checksums. + Only processes .md files, logs and skips others. + + Args: + directory: Directory to scan + + Returns: + ScanResult containing found files and any errors + """ + logger.debug(f"Scanning directory: {directory}") + result = ScanResult() + + if not directory.exists(): + logger.debug(f"Directory does not exist: {directory}") + return result + + for path in directory.rglob("*"): + if not path.is_file() or not path.name.endswith(".md"): + if path.is_file(): + logger.debug(f"Skipping non-markdown file: {path}") + continue + + try: + # Get relative path first - used in error reporting if needed + rel_path = str(path.relative_to(directory)) + checksum = await self.file_service.compute_checksum(rel_path) + result.files[rel_path] = checksum + result.checksums[checksum] = rel_path + + except Exception as e: + rel_path = str(path.relative_to(directory)) + result.errors[rel_path] = str(e) + logger.error(f"Failed to read {rel_path}: {e}") + + logger.debug(f"Found {len(result.files)} markdown files") + if result.errors: + logger.warning(f"Encountered {len(result.errors)} errors while scanning") + + return result diff --git a/src/basic_memory/sync/sync_service.py.bak b/src/basic_memory/sync/sync_service_old.py similarity index 100% rename from src/basic_memory/sync/sync_service.py.bak rename to src/basic_memory/sync/sync_service_old.py diff --git a/tests/conftest.py b/tests/conftest.py index fe919995..2e866fb2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -322,4 +322,4 @@ async def test_graph( @pytest_asyncio.fixture def watch_service(sync_service, file_service, test_config): - return WatchService(sync_service=sync_service, file_service=file_service, config=test_config) + return WatchService(sync_service=sync_service, file_service=file_service, config=test_config) \ No newline at end of file diff --git a/tests/sync/test_sync_service.py b/tests/sync/test_sync_service.py index ddd9631d..11677bb7 100644 --- a/tests/sync/test_sync_service.py +++ b/tests/sync/test_sync_service.py @@ -48,7 +48,7 @@ type: knowledge # Verify forward reference source = await entity_service.get_by_permalink("source") - assert len(source.relations) == 2 + assert len(source.relations) == 1 assert source.relations[0].to_id is None assert source.relations[0].to_name == "target-doc" @@ -68,7 +68,7 @@ Target content # Verify reference is now resolved source = await entity_service.get_by_permalink("source") target = await entity_service.get_by_permalink("target-doc") - assert len(source.relations) == 2 + assert len(source.relations) == 1 assert source.relations[0].to_id == target.id assert source.relations[0].to_name == target.title @@ -130,7 +130,7 @@ A test concept. # with forward link entity = await entity_service.get_by_permalink(test_concept.permalink) relations = entity.relations - assert len(relations) == 1 + assert len(relations) == 1, "Expected 1 relation for entity" assert relations[0].to_name == "concept/other" @@ -470,8 +470,12 @@ modified: 2024-01-01 # Verify final state doc = await sync_service.entity_service.repository.get_by_permalink("changing") assert doc is not None - # File should have a checksum, even if it's from either version - assert doc.checksum is not None + + # if we failed in the middle of a sync, the next one should fix it. + if doc.checksum is None: + await sync_service.sync(test_config.home) + doc = await sync_service.entity_service.repository.get_by_permalink("changing") + assert doc.checksum is not None @pytest.mark.asyncio @@ -529,7 +533,7 @@ async def test_handle_entity_deletion( root_entity = test_graph["root"] # Delete the entity - await sync_service.handle_entity_deletion(root_entity.file_path) + await sync_service.handle_delete(root_entity.file_path) # Verify entity is gone from db assert await entity_repository.get_by_permalink(root_entity.permalink) is None