feat: process fork repos only if their parent repo were already processed

Author: mbani01

services/apps/git_integration/src/crowdgit/database/crud.py

@@ -36,6 +36,28 @@ async def get_repository_by_url(url: str) -> dict[str, Any] | None:
     return dict(result) if result else None
 
 
+async def get_recently_processed_repository_by_url(url: str) -> Repository | None:
+    """
+    Get repository by URL that was processed within the configured update interval.
+
+    Returns the repository only if it was last processed within REPOSITORY_UPDATE_INTERVAL_HOURS
+    and has a COMPLETED state.
+    Used to check if a repository needs reprocessing based on the update interval.
+    """
+    sql_query = """
+    SELECT id, url, state, priority, "lastProcessedAt", "lockedAt", "createdAt", "updatedAt", "maintainerFile", "forkedFrom"
+    FROM git.repositories
+    WHERE url = $1
+        AND "deletedAt" IS NULL
+        AND "lastProcessedAt" > NOW() - INTERVAL '1 hour' * $2
+        AND state = $3
+    """
+    result = await fetchrow(
+        sql_query, (url, REPOSITORY_UPDATE_INTERVAL_HOURS, RepositoryState.COMPLETED)
+    )
+    return Repository.from_db(dict(result)) if result else None
+
+
 async def acquire_onboarding_repo() -> Repository | None:
     onboarding_repo_sql_query = """
     WITH current_onboarding_count AS (

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

@@ -28,6 +28,7 @@ class RepositoryState(str, Enum):
     PROCESSING = "processing"
     COMPLETED = "completed"
     FAILED = "failed"
+    REQUIRES_PARENT = "requires_parent"  # fork repo without valid parent repo in out system
 
 
 class RepositoryPriority(int):

services/apps/git_integration/src/crowdgit/models/repository.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import uuid
 from datetime import datetime
 from typing import Any
@@ -36,11 +38,14 @@ class Repository(BaseModel):
         None,
         description="The source repository URL if this repository is a fork",
     )
+    parent_repo: Repository | None = Field(
+        None, description="The parent repository (in case of fork) object from our database"
+    )
     created_at: datetime = Field(..., description="Creation timestamp")
     updated_at: datetime = Field(..., description="Last update timestamp")
 
     @classmethod
-    def from_db(cls, db_data: dict[str, Any]) -> "Repository":
+    def from_db(cls, db_data: dict[str, Any]) -> Repository:
         """Create Repository instance from database data"""
         # Convert database field names to model field names
         repo_data = db_data.copy()

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

@@ -2,6 +2,7 @@
 
 from crowdgit.database.crud import (
     acquire_repo_for_processing,
+    get_recently_processed_repository_by_url,
     mark_repo_as_processed,
     release_repo,
     update_last_processed_commit,
@@ -135,14 +136,41 @@ def _reset_all_contexts(self) -> None:
         for service in services:
             service.reset_logger_context()
 
+    async def _check_parent_repo_validity(self, repository: Repository) -> bool:
+        """
+        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
+        """
+        if not repository.forked_from:
+            return True
+        logger.info(
+            f"Repo forked from {repository.forked_from}, checking parent repo validity in our system"
+        )
+        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
+
     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.PENDING
+        processing_state = RepositoryState.FAILED
 
         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
             async for batch_info in self.clone_service.clone_batches_generator(
                 repository,
                 working_dir_cleanup=True,
@@ -166,7 +194,6 @@ async def _process_single_repository(self, repository: Repository):
             logger.info("Incremental processing completed successfully")
             processing_state = RepositoryState.COMPLETED
         except Exception as e:
-            processing_state = RepositoryState.FAILED
             logger.error(f"Processing failed with error: {repr(e)}")
         finally:
             # Reset logger context for all services