feat: skip activities from parent repo in case of fork

Author: mbani01

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

@@ -45,7 +45,7 @@ async def get_recently_processed_repository_by_url(url: str) -> Repository | Non
     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"
+    SELECT id, url, state, priority, "lastProcessedAt", "lockedAt", "createdAt", "updatedAt", "maintainerFile", "forkedFrom", "segmentId"
     FROM git.repositories
     WHERE url = $1
         AND "deletedAt" IS NULL
@@ -305,6 +305,56 @@ async def set_maintainer_end_date(
     )
 
 
+async def batch_check_parent_activities(
+    activity_keys: list[tuple[str, str, str]],
+    parent_channel: str,
+    parent_segment_id: str,
+) -> set[str]:
+    """
+    Batch check which activities exist in parent repo using full dedup key.
+
+    Args:
+        activity_keys: List of (timestamp, type, sourceId) tuples
+        parent_channel: Parent repository URL
+        parent_segment_id: Parent repository segment ID
+
+    Returns:
+        Set of sourceIds that exist in parent repo
+    """
+    if not activity_keys:
+        return set()
+
+    # Use dedup index with ALL fields for optimal performance
+    # Index: (timestamp, platform, type, sourceId, channel, segmentId)
+    # Build OR conditions for each (timestamp, type, sourceId) combination
+    conditions = []
+    params = ["git", parent_channel, parent_segment_id]
+    param_idx = 4
+
+    for timestamp_str, activity_type, source_id in activity_keys:
+        conditions.append(
+            f'("timestamp" = ${param_idx} AND "type" = ${param_idx + 1} AND "sourceId" = ${param_idx + 2})'
+        )
+        timestamp = datetime.fromisoformat(timestamp_str)
+        params.append(timestamp)
+        params.append(activity_type)
+        params.append(source_id)
+        param_idx += 3
+
+    sql_query = f"""
+    SELECT DISTINCT "sourceId"
+    FROM "activityRelations"
+    WHERE "platform" = $1
+        AND "channel" = $2
+        AND "segmentId" = $3
+        AND ({" OR ".join(conditions)})
+    """
+
+    result = await query(sql_query, tuple(params))
+
+    return {row["sourceId"] for row in result}
+
+
 async def save_service_execution(service_execution: ServiceExecution) -> None:
     """
     Save service execution record to database.

services/apps/git_integration/src/crowdgit/services/commit/commit_service.py

@@ -13,7 +13,11 @@
 from pydantic import validate_email
 from tenacity import retry, stop_after_attempt, wait_fixed
 
-from crowdgit.database.crud import batch_insert_activities, save_service_execution
+from crowdgit.database.crud import (
+    batch_check_parent_activities,
+    batch_insert_activities,
+    save_service_execution,
+)
 from crowdgit.enums import (
     DataSinkWorkerQueueMessageType,
     ErrorCode,
@@ -115,6 +119,7 @@ async def process_single_batch_commits(
                 "total_commits": 0,
                 "processed_commits": 0,
                 "bad_commits": 0,
+                "skipped_activities": 0,
                 "total_activities": 0,
             }
 
@@ -139,6 +144,7 @@ async def process_single_batch_commits(
                 batch_info.remote,
                 repository.segment_id,
                 repository.integration_id,
+                repository.parent_repo,
             )
 
             batch_end_time = time.time()
@@ -164,6 +170,7 @@ async def process_single_batch_commits(
                         "total_commits": self._metrics_context["total_commits"],
                         "processed_commits": self._metrics_context["processed_commits"],
                         "bad_commits": self._metrics_context["bad_commits"],
+                        "skipped_activities": self._metrics_context["skipped_activities"],
                         "total_activities": self._metrics_context["total_activities"],
                     },
                 )
@@ -200,6 +207,7 @@ async def process_single_batch_commits(
                     "total_commits": self._metrics_context["total_commits"],
                     "processed_commits": self._metrics_context["processed_commits"],
                     "bad_commits": self._metrics_context["bad_commits"],
+                    "skipped_activities": self._metrics_context["skipped_activities"],
                     "total_activities": self._metrics_context["total_activities"],
                 },
             )
@@ -609,6 +617,66 @@ def create_activities_from_commit(
 
         return activities_db, activities_queue
 
+    async def _filter_parent_repo_activities(
+        self,
+        activities_db: list[tuple],
+        activities_queue: list[dict],
+        parent_repo: Repository,
+    ) -> tuple[list[tuple], list[dict], int]:
+        """
+        Filter out activities that exist in parent repo (for fork detection).
+        Uses full dedup key (timestamp, platform, type, sourceId, channel, segmentId) for optimal index usage.
+
+        Args:
+            activities_db: List of activity tuples for database
+            activities_queue: List of activity dicts for Kafka queue
+            parent_repo: Parent repository information
+
+        Returns:
+            Tuple of (filtered_activities_db, filtered_activities_queue, skipped_activities_count)
+        """
+        if not activities_db:
+            return activities_db, activities_queue, 0
+
+        # Extract (timestamp, type, sourceId) for each activity to use full dedup index
+        activity_keys = []
+        for act in activities_db:
+            data = orjson.loads(act[2])["data"]
+            activity_keys.append((data["timestamp"], data["type"], data["sourceId"]))
+
+        # Batch check which activities exist in parent repo
+        parent_source_ids = await batch_check_parent_activities(
+            activity_keys,
+            parent_repo.url,
+            parent_repo.segment_id,
+        )
+
+        if not parent_source_ids:
+            return activities_db, activities_queue, 0
+
+        filtered_activities_db = []
+        filtered_activities_queue = []
+        skipped_activities_count = 0
+
+        for i, activity_tuple in enumerate(activities_db):
+            activity_data = orjson.loads(activity_tuple[2])
+            source_id = activity_data["data"]["sourceId"]
+
+            if source_id not in parent_source_ids:
+                # Activity doesn't exist in parent repo, keep it
+                filtered_activities_db.append(activity_tuple)
+                filtered_activities_queue.append(activities_queue[i])
+            else:
+                # Activity exists in parent repo, skip it
+                skipped_activities_count += 1
+
+        if skipped_activities_count > 0:
+            self.logger.info(
+                f"Filtered out {skipped_activities_count} activities from parent repo {parent_repo.url}"
+            )
+
+        return filtered_activities_db, filtered_activities_queue, skipped_activities_count
+
     async def process_commits_chunk(
         self,
         commit_texts_chunk: list[str | None],
@@ -617,6 +685,7 @@ async def process_commits_chunk(
         remote: str,
         segment_id: str,
         integration_id: str,
+        parent_repo: Repository | None,
     ) -> None:
         """
         Process a chunk of raw commit texts into activities and write them to DB and Kafka.
@@ -674,15 +743,31 @@ async def process_commits_chunk(
                 del commit_lines
                 del numstats_text
 
-        self.logger.info(
-            f"Processed {processed_commits} commits, skipped {bad_commits} invalid commits in {repo_path}"
-        )
+        # Filter out activities from parent repo (for forks)
+        skipped_activities = 0
+        if parent_repo:
+            (
+                activities_db,
+                activities_queue,
+                skipped_activities,
+            ) = await self._filter_parent_repo_activities(
+                activities_db, activities_queue, parent_repo
+            )
 
+        if skipped_activities > 0:
+            self.logger.info(
+                f"Processed {processed_commits} commits, skipped {bad_commits} invalid commits, filtered {skipped_activities} activities from parent repo in {repo_path}"
+            )
+        else:
+            self.logger.info(
+                f"Processed {processed_commits} commits, skipped {bad_commits} invalid commits in {repo_path}"
+            )
         # Update metrics context
         if self._metrics_context:
             self._metrics_context["processed_commits"] += processed_commits
             self._metrics_context["bad_commits"] += bad_commits
             self._metrics_context["total_activities"] += len(activities_db)
+            self._metrics_context["skipped_activities"] += skipped_activities
 
         # Write activities to database and queue
         if activities_db:
@@ -701,6 +786,7 @@ async def _process_activities_from_commits(
         remote: str,
         segment_id: str,
         integration_id: str,
+        parent_repo: Repository | None = None,
     ):
         """
         Parse raw git log output, process commits into activities, and save to database.
@@ -747,6 +833,7 @@ async def process_single_chunk(chunk_start_idx: int, chunk_end_idx: int):
                         remote,
                         segment_id,
                         integration_id,
+                        parent_repo,
                     )
                     completed_chunks += 1
                     self.logger.info(f"Progress: {completed_chunks}/{total_chunks} chunks")