chore: refactore _validate_and_get_parent_repo

Author: mbani01

services/apps/git_integration/src/crowdgit/enums.py

@@ -19,6 +19,7 @@ class ErrorCode(str, Enum):
     MAINTAINER_ANALYSIS_FAILED = "maintainer-analysis-failed"
     MAINTAINER_INTERVAL_NOT_ELAPSED = "maintainer-interval-not-elapsed"
     CLEANUP_FAILED = "cleanup-failed"
+    PARENT_REPO_INVALID = "parent-repo-invalid"
 
 
 class RepositoryState(str, Enum):

services/apps/git_integration/src/crowdgit/errors.py

@@ -101,3 +101,9 @@ class MaintainerIntervalNotElapsedError(CrowdGitError):
     error_message: str = "Maintainer processing interval has not elapsed yet"
     error_code: ErrorCode = ErrorCode.MAINTAINER_INTERVAL_NOT_ELAPSED
     ai_cost: int = 0
+
+
+@dataclass
+class ParentRepoInvalidError(CrowdGitError):
+    error_message: str = "Parent repository is not valid or not found"
+    error_code: ErrorCode = ErrorCode.PARENT_REPO_INVALID

services/apps/git_integration/src/crowdgit/worker/repository_worker.py

@@ -8,7 +8,7 @@
     update_last_processed_commit,
 )
 from crowdgit.enums import RepositoryState
-from crowdgit.errors import InternalError
+from crowdgit.errors import InternalError, ParentRepoInvalidError
 
 # Import configured loguru logger from crowdgit.logger
 from crowdgit.logger import logger
@@ -136,41 +136,51 @@ def _reset_all_contexts(self) -> None:
         for service in services:
             service.reset_logger_context()
 
-    async def _check_parent_repo_validity(self, repository: Repository) -> bool:
+    async def _validate_and_get_parent_repo(self, repository: Repository) -> Repository | None:
         """
-        In case of forked repo we need to prevent re-processing activities from parent repo and assigning them to fork, so we need to check:
-            1. Parent repo already connected/onboarded
-            2. Parent repo was processed successfully from last run to ensure we have up to date data
-        also assigns repository.parent_repo  if valid
+        Validate and return the parent repository for forks.
+        For forked repos, we need to prevent re-processing activities from the parent repo,
+        so we verify the parent:
+            1. Is already connected/onboarded in our system
+            2. Was processed successfully within REPOSITORY_UPDATE_INTERVAL_HOURS
+            3. Has COMPLETED state
+
+        Returns:
+            Repository | None: Parent repository if this is a valid fork, None if not a fork
+        Raises:
+            ParentRepoInvalidError: If this is a fork but the parent repo is invalid or not found
         """
         if not repository.forked_from:
-            return True
+            return None
+
         logger.info(
-            f"Repo forked from {repository.forked_from}, checking parent repo validity in our system"
+            f"Repository {repository.url} is forked from {repository.forked_from}, validating parent repo..."
         )
-        await asyncio.sleep(10)
+
         parent_repo = await get_recently_processed_repository_by_url(repository.forked_from)
         if not parent_repo:
             logger.warning(
                 f"Parent repo {repository.forked_from} is not found/valid - Aborting processing"
             )
-            return False
-        logger.info("Parent repo is valid and already processed, proceeding with fork processing")
-        repository.parent_repo = parent_repo
-        return True
+            raise ParentRepoInvalidError()
+
+        logger.info(
+            f"Parent repo {repository.forked_from} is valid, proceeding with fork processing"
+        )
+        return parent_repo
 
     async def _process_single_repository(self, repository: Repository):
         """Process a single repository through services with full clone for new repos, incremental for existing"""
         logger.info("Processing repository: {}", repository.url)
-        processing_state = RepositoryState.FAILED
+        processing_state = RepositoryState.PENDING
 
         try:
             repo_name = get_repo_name(repository.url)
             self._bind_repository_context(repository, repo_name)
-            valid_parent = await self._check_parent_repo_validity(repository)
-            if not valid_parent:
-                processing_state = RepositoryState.REQUIRES_PARENT
-                return
+
+            # Validate and get parent repo if this is a fork
+            repository.parent_repo = await self._validate_and_get_parent_repo(repository)
+
             async for batch_info in self.clone_service.clone_batches_generator(
                 repository,
                 working_dir_cleanup=True,
@@ -193,8 +203,12 @@ async def _process_single_repository(self, repository: Repository):
 
             logger.info("Incremental processing completed successfully")
             processing_state = RepositoryState.COMPLETED
+        except ParentRepoInvalidError as e:
+            logger.error(f"Parent repo validation failed: {repr(e)}")
+            processing_state = RepositoryState.REQUIRES_PARENT
         except Exception as e:
             logger.error(f"Processing failed with error: {repr(e)}")
+            processing_state = RepositoryState.FAILED
         finally:
             # Reset logger context for all services
             self._reset_all_contexts()