[ENG-1404] Duplicate DSR - runner integration

CHANGELOG.md

@@ -24,6 +24,7 @@ Changes can also be flagged with a GitHub label for tracking purposes. The URL o
 ### Added
 - Added a duplicate group table with deterministic ids [#6881](https://github.com/ethyca/fides/pull/6881) https://github.com/ethyca/fides/labels/db-migration
 - Added replace mode for decoding context loggers to avoid decode errors with zip files[#6899](https://github.com/ethyca/fides/pull/6899/files)
+- Added duplicate DSR checking to request runner [#6860](https://github.com/ethyca/fides/pull/6860/)
 
 ### Changed
 - Updated filter modal in new privacy request screen to store filters as query params in url [#6818](https://github.com/ethyca/fides/pull/6818)

src/fides/api/service/privacy_request/duplication_detection.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timedelta, timezone
 from typing import Optional
+from uuid import UUID
 
 from loguru import logger
 from sqlalchemy.orm import Session
@@ -21,6 +22,7 @@
 from fides.api.task.conditional_dependencies.sql_translator import (
     SQLConditionTranslator,
 )
+from fides.config.config_proxy import ConfigProxy
 from fides.config.duplicate_detection_settings import DuplicateDetectionSettings
 
 ACTIONED_REQUEST_STATUSES = [
@@ -38,9 +40,10 @@
 class DuplicateDetectionService:
     def __init__(self, db: Session):
         self.db = db
+        self.config = ConfigProxy(db).privacy_request_duplicate_detection
 
     def _create_identity_conditions(
-        self, current_request: PrivacyRequest, config: DuplicateDetectionSettings
+        self, current_request: PrivacyRequest
     ) -> list[Condition]:
         """Creates conditions for matching identity fields.
 
@@ -52,12 +55,12 @@ def _create_identity_conditions(
         current_identities: dict[str, str] = {
             pi.field_name: pi.hashed_value
             for pi in current_request.provided_identities  # type: ignore [attr-defined]
-            if pi.field_name in config.match_identity_fields
+            if pi.field_name in self.config.match_identity_fields
         }
-        if len(current_identities) != len(config.match_identity_fields):
+        if len(current_identities) != len(self.config.match_identity_fields):
             missing_fields = [
                 field
-                for field in config.match_identity_fields
+                for field in self.config.match_identity_fields
                 if field not in current_identities.keys()
             ]
             logger.debug(
@@ -103,7 +106,6 @@ def _create_time_window_condition(self, time_window_days: int) -> Condition:
     def create_duplicate_detection_conditions(
         self,
         current_request: PrivacyRequest,
-        config: DuplicateDetectionSettings,
     ) -> Optional[ConditionGroup]:
         """
         Create conditions for duplicate detection based on configuration.
@@ -115,15 +117,15 @@ def create_duplicate_detection_conditions(
         Returns:
             A ConditionGroup with AND operator, or None if no conditions can be created
         """
-        if len(config.match_identity_fields) == 0:
+        if len(self.config.match_identity_fields) == 0:
             return None
 
-        identity_conditions = self._create_identity_conditions(current_request, config)
+        identity_conditions = self._create_identity_conditions(current_request)
         if not identity_conditions:
             return None  # Only proceed if we have identity conditions
 
         time_window_condition = self._create_time_window_condition(
-            config.time_window_days
+            self.config.time_window_days
         )
 
         # Combine all conditions with AND operator
@@ -135,7 +137,6 @@ def create_duplicate_detection_conditions(
     def find_duplicate_privacy_requests(
         self,
         current_request: PrivacyRequest,
-        config: DuplicateDetectionSettings,
     ) -> list[PrivacyRequest]:
         """
         Find potential duplicate privacy requests based on duplicate detection configuration.
@@ -151,7 +152,7 @@ def find_duplicate_privacy_requests(
             List of PrivacyRequest objects that match the duplicate criteria,
             does not include the current request
         """
-        condition = self.create_duplicate_detection_conditions(current_request, config)
+        condition = self.create_duplicate_detection_conditions(current_request)
 
         if condition is None:
             return []
@@ -162,28 +163,75 @@ def find_duplicate_privacy_requests(
         query = query.filter(PrivacyRequest.id != current_request.id)
         return query.all()
 
-    def generate_dedup_key(
-        self, request: PrivacyRequest, config: DuplicateDetectionSettings
-    ) -> str:
+    def generate_dedup_key(self, request: PrivacyRequest) -> str:
         """
         Generate a dedup key for a request based on the duplicate detection settings.
         """
         current_identities: dict[str, str] = {
             pi.field_name: pi.hashed_value
             for pi in request.provided_identities  # type: ignore [attr-defined]
-            if pi.field_name in config.match_identity_fields
+            if pi.field_name in self.config.match_identity_fields
         }
-        if len(current_identities) != len(config.match_identity_fields):
+        if len(current_identities) != len(self.config.match_identity_fields):
             raise ValueError(
                 "This request does not contain the required identity fields for duplicate detection."
             )
         return "|".join(
             [
                 current_identities[field]
-                for field in sorted(config.match_identity_fields)
+                for field in sorted(self.config.match_identity_fields)
             ]
         )
 
+    def update_duplicate_group_ids(
+        self,
+        request: PrivacyRequest,
+        duplicates: list[PrivacyRequest],
+        duplicate_group_id: UUID,
+    ) -> None:
+        """
+        Update the duplicate request group ids for a request and its duplicates.
+        Args:
+            request: The privacy request to update
+            duplicates: The list of duplicate requests to update
+            duplicate_group_id: The duplicate request group id to update
+        """
+        update_all = [request] + duplicates
+        try:
+            for privacy_request in update_all:
+                privacy_request.duplicate_request_group_id = duplicate_group_id  # type: ignore [assignment]
+        except Exception as e:
+            logger.error(f"Failed to update duplicate request group ids: {e}")
+            raise e
+
+    def add_error_execution_log(self, request: PrivacyRequest, message: str) -> None:
+        request.add_error_execution_log(
+            db=self.db,
+            connection_key=None,
+            dataset_name="Duplicate Request Detection",
+            collection_name=None,
+            message=message,
+            action_type=(
+                request.policy.get_action_type()  # type: ignore [arg-type]
+                if request.policy
+                else ActionType.access
+            ),
+        )
+
+    def add_success_execution_log(self, request: PrivacyRequest, message: str) -> None:
+        request.add_success_execution_log(
+            db=self.db,
+            connection_key=None,
+            dataset_name="Duplicate Request Detection",
+            collection_name=None,
+            message=message,
+            action_type=(
+                request.policy.get_action_type()  # type: ignore [arg-type]
+                if request.policy
+                else ActionType.access
+            ),
+        )
+
     def verified_identity_cases(
         self, request: PrivacyRequest, duplicates: list[PrivacyRequest]
     ) -> bool:
@@ -206,60 +254,52 @@ def verified_identity_cases(
         # The request identity is not verified.
         if not request.identity_verified_at:
             if len(verified_in_group) > 0:
-                logger.debug(
-                    f"Request {request.id} is a duplicate: it is not verified and duplicating verified request(s) {verified_in_group}."
-                )
+                message = f"Request {request.id} is a duplicate: it is duplicating request(s) {[duplicate.id for duplicate in verified_in_group]}."
+                logger.debug(message)
+                self.add_error_execution_log(request, message)
                 return True
 
-            min_created_at = min(
-                (d.created_at for d in duplicates if d.created_at), default=None
-            ) or datetime.now(timezone.utc)
-            request_created_at = (
-                request.created_at
-                if request.created_at is not None
-                else datetime.now(timezone.utc)
+            canonical_request = min(duplicates, key=lambda x: x.created_at)  # type: ignore [arg-type, return-value]
+            canonical_request_created_at = canonical_request.created_at or datetime.now(
+                timezone.utc
             )
-            if request_created_at < min_created_at:
-                logger.debug(
-                    f"Request {request.id} is not a duplicate: it is the first request to be created in the group."
-                )
+            request_created_at = request.created_at or datetime.now(timezone.utc)
+            if request_created_at < canonical_request_created_at:
+                message = f"Request {request.id} is not a duplicate: it is the first request to be created in the group."
+                logger.debug(message)
+                self.add_success_execution_log(request, message)
                 return False
-            logger.debug(
-                f"Request {request.id} is a duplicate: it is not verified and is not the first request to be created in the group."
-            )
+
+            message = f"Request {request.id} is a duplicate: it is duplicating request(s) ['{canonical_request.id}']."
+            logger.debug(message)
+            self.add_error_execution_log(request, message)
             return True
 
         # The request identity is verified.
         if not verified_in_group:
-            logger.debug(
-                f"Request {request.id} is not a duplicate: it is verified and no other requests in the group are verified."
-            )
+            message = f"Request {request.id} is not a duplicate: it is the first request to be verified in the group."
+            logger.debug(message)
+            self.add_success_execution_log(request, message)
             return False
 
         # If this request is the first with a verified identity, it is not a duplicate.
-        min_verified_at = min(
-            (d.identity_verified_at for d in duplicates if d.identity_verified_at),
-            default=None,
-        ) or datetime.now(timezone.utc)
-        request_verified_at = (
-            request.identity_verified_at
-            if request.identity_verified_at is not None
-            else datetime.now(timezone.utc)
+        canonical_request = min(duplicates, key=lambda x: x.identity_verified_at)  # type: ignore [arg-type, return-value]
+        canonical_request_verified_at = (
+            canonical_request.identity_verified_at or datetime.now(timezone.utc)
         )
-        if request_verified_at < min_verified_at:
-            logger.debug(
-                f"Request {request.id} is not a duplicate: it is the first request to be verified in the group."
-            )
+        request_verified_at = request.identity_verified_at or datetime.now(timezone.utc)
+        if request_verified_at < canonical_request_verified_at:
+            message = f"Request {request.id} is not a duplicate: it is the first request to be verified in the group."
+            logger.debug(message)
+            self.add_success_execution_log(request, message)
             return False
-        logger.debug(
-            f"Request {request.id} is a duplicate: it is verified but not the first request to be verified in the group."
-        )
+        message = f"Request {request.id} is a duplicate: it is duplicating request(s) ['{canonical_request.id}']."
+        logger.debug(message)
+        self.add_error_execution_log(request, message)
         return True
 
     # pylint: disable=too-many-return-statements
-    def is_duplicate_request(
-        self, request: PrivacyRequest, config: DuplicateDetectionSettings
-    ) -> bool:
+    def is_duplicate_request(self, request: PrivacyRequest) -> bool:
         """
         Determine if a request is a duplicate request and assigns a duplicate request group id.
 
@@ -281,52 +321,44 @@ def is_duplicate_request(
         Returns:
             True if the request is a duplicate request, False otherwise
         """
-        duplicates = self.find_duplicate_privacy_requests(request, config)
-        rule_version = generate_rule_version(config)
+        duplicates = self.find_duplicate_privacy_requests(request)
+        rule_version = generate_rule_version(
+            DuplicateDetectionSettings(
+                enabled=self.config.enabled,
+                time_window_days=self.config.time_window_days,
+                match_identity_fields=self.config.match_identity_fields,
+            )
+        )
         try:
-            dedup_key = self.generate_dedup_key(request, config)
+            dedup_key = self.generate_dedup_key(request)
         except ValueError as e:
-            logger.debug(f"Request {request.id} is not a duplicate: {e}")
+            message = f"Request {request.id} is not a duplicate: {e}"
+            logger.debug(message)
+            self.add_success_execution_log(request, message)
             return False
 
         _, duplicate_group = DuplicateGroup.get_or_create(
             db=self.db, data={"rule_version": rule_version, "dedup_key": dedup_key}
         )
         if duplicate_group is None:
-            logger.error(
-                f"Failed to create duplicate group for request {request.id} with dedup key {dedup_key}"
-            )
+            message = f"Failed to create duplicate group for request {request.id} with dedup key {dedup_key}"
+            logger.error(message)
+            self.add_error_execution_log(request, message)
             return False
-        logger.info(
-            f"Duplicate group {duplicate_group.id} created for request {request.id} with dedup key {dedup_key}"
-        )
-        request.update(
-            db=self.db, data={"duplicate_request_group_id": duplicate_group.id}
-        )
+
+        self.update_duplicate_group_ids(request, duplicates, duplicate_group.id)  # type: ignore [arg-type]
 
         # if this is the only request in the group, it is not a duplicate
         if len(duplicates) == 0:
-            logger.debug(
-                f"Request {request.id} is not a duplicate: no matching requests were found."
-            )
+            message = f"Request {request.id} is not a duplicate."
+            logger.debug(message)
+            self.add_success_execution_log(request, message)
             return False
 
         if request.status == PrivacyRequestStatus.duplicate:
-            logger.warning(
-                f"Request {request.id} is a duplicate request that was requeued. This should not happen."
-            )
-            request.add_error_execution_log(
-                db=self.db,
-                connection_key=None,
-                dataset_name="Duplicate Request Detection",
-                collection_name=None,
-                message=f"Request {request.id} is a duplicate request that was requeued. This should not happen.",
-                action_type=(
-                    request.policy.get_action_type()  # type: ignore [arg-type]
-                    if request.policy
-                    else ActionType.access
-                ),
-            )
+            message = f"Request {request.id} is a duplicate request that was requeued. This should not happen."
+            logger.warning(message)
+            self.add_error_execution_log(request, message)
             return True
 
         # only compare to non-duplicate requests for the following cases
@@ -337,9 +369,9 @@ def is_duplicate_request(
         ]
         # If no non-duplicate requests are found, this request is not a duplicate.
         if len(canonical_requests) == 0:
-            logger.debug(
-                f"Request {request.id} is not a duplicate: all matching requests have been marked as duplicate requests."
-            )
+            message = f"Request {request.id} is not a duplicate."
+            logger.debug(message)
+            self.add_success_execution_log(request, message)
             return False
 
         # If any requests in group are actioned, this request is a duplicate.
@@ -349,9 +381,9 @@ def is_duplicate_request(
             if duplicate.status in ACTIONED_REQUEST_STATUSES
         ]
         if len(actioned_in_group) > 0:
-            logger.debug(
-                f"Request {request.id} is a duplicate: it is duplicating actioned request(s) {actioned_in_group}."
-            )
+            message = f"Request {request.id} is a duplicate: it is duplicating actioned request(s) {[duplicate.id for duplicate in actioned_in_group]}."
+            logger.debug(message)
+            self.add_error_execution_log(request, message)
             return True
         # Check against verified identity rules.
         return self.verified_identity_cases(request, canonical_requests)

src/fides/api/service/privacy_request/request_runner_service.py

@@ -71,6 +71,9 @@
     get_attachments_content,
     process_attachments_for_upload,
 )
+from fides.api.service.privacy_request.duplication_detection import (
+    DuplicateDetectionService,
+)
 from fides.api.service.storage.storage_uploader_service import upload
 from fides.api.task.filter_results import filter_data_categories
 from fides.api.task.graph_runners import access_runner, consent_runner, erasure_runner
@@ -446,6 +449,19 @@ def run_privacy_request(
                 logger.info("Terminating privacy request: request deleted.")
                 return
 
+            duplicate_detection_service = DuplicateDetectionService(session)
+            # Service initializes with ConfigProxy, so we can check if duplicate detection is enabled
+            if duplicate_detection_service.config.enabled:
+                logger.info(
+                    "Duplicate detection is enabled. Checking if privacy request is a duplicate."
+                )
+                if duplicate_detection_service.is_duplicate_request(privacy_request):
+                    logger.info("Terminating privacy request: request is a duplicate.")
+                    privacy_request.update(
+                        session, data={"status": PrivacyRequestStatus.duplicate}
+                    )
+                    return
+
             logger.info("Dispatching privacy request")
             privacy_request.start_processing(session)