[ENG-1404] Duplicate DSR - runner integration

CHANGELOG.md

@@ -23,6 +23,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 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/models/privacy_request/duplicate_group.py

@@ -1,7 +1,7 @@
 import hashlib
 import json
 import uuid
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Union
 
 from sqlalchemy import Boolean, Column, Text
 from sqlalchemy.dialects.postgresql import UUID
@@ -31,7 +31,7 @@ def generate_deterministic_uuid(rule_version: str, dedup_key: str) -> uuid.UUID:
     return uuid.UUID(hashlib.md5(hash_input).hexdigest())
 
 
-def generate_rule_version(config: DuplicateDetectionSettings) -> str:
+def generate_rule_version(config: Union[DuplicateDetectionSettings]) -> str:
     """Generate a stable short hash for the dedup rule config."""
     normalized = json.dumps(config.model_dump(), sort_keys=True)
     return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:8]

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 DuplicateDetectionSettingsProxy
 from fides.config.duplicate_detection_settings import DuplicateDetectionSettings
 
 ACTIONED_REQUEST_STATUSES = [
@@ -40,7 +42,7 @@ def __init__(self, db: Session):
         self.db = db
 
     def _create_identity_conditions(
-        self, current_request: PrivacyRequest, config: DuplicateDetectionSettings
+        self, current_request: PrivacyRequest, config: DuplicateDetectionSettingsProxy
     ) -> list[Condition]:
         """Creates conditions for matching identity fields.
 
@@ -103,7 +105,7 @@ def _create_time_window_condition(self, time_window_days: int) -> Condition:
     def create_duplicate_detection_conditions(
         self,
         current_request: PrivacyRequest,
-        config: DuplicateDetectionSettings,
+        config: DuplicateDetectionSettingsProxy,
     ) -> Optional[ConditionGroup]:
         """
         Create conditions for duplicate detection based on configuration.
@@ -135,7 +137,7 @@ def create_duplicate_detection_conditions(
     def find_duplicate_privacy_requests(
         self,
         current_request: PrivacyRequest,
-        config: DuplicateDetectionSettings,
+        config: DuplicateDetectionSettingsProxy,
     ) -> list[PrivacyRequest]:
         """
         Find potential duplicate privacy requests based on duplicate detection configuration.
@@ -158,12 +160,11 @@ def find_duplicate_privacy_requests(
 
         translator = SQLConditionTranslator(self.db)
         query = translator.generate_query_from_condition(condition)
-
         query = query.filter(PrivacyRequest.id != current_request.id)
         return query.all()
 
     def generate_dedup_key(
-        self, request: PrivacyRequest, config: DuplicateDetectionSettings
+        self, request: PrivacyRequest, config: DuplicateDetectionSettingsProxy
     ) -> str:
         """
         Generate a dedup key for a request based on the duplicate detection settings.
@@ -184,6 +185,55 @@ def generate_dedup_key(
             ]
         )
 
+    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,59 +256,53 @@ 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
+        self, request: PrivacyRequest, config: DuplicateDetectionSettingsProxy
     ) -> bool:
         """
         Determine if a request is a duplicate request and assigns a duplicate request group id.
@@ -282,51 +326,43 @@ def is_duplicate_request(
             True if the request is a duplicate request, False otherwise
         """
         duplicates = self.find_duplicate_privacy_requests(request, config)
-        rule_version = generate_rule_version(config)
+        rule_version = generate_rule_version(
+            DuplicateDetectionSettings(
+                enabled=config.enabled,
+                time_window_days=config.time_window_days,
+                match_identity_fields=config.match_identity_fields,
+            )
+        )
         try:
             dedup_key = self.generate_dedup_key(request, config)
         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 +373,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 +385,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,21 @@ def run_privacy_request(
                 logger.info("Terminating privacy request: request deleted.")
                 return
 
+            if ConfigProxy(session).privacy_request_duplicate_detection.enabled:
+                logger.info(
+                    "Duplicate detection is enabled. Checking if privacy request is a duplicate."
+                )
+                duplicate_detection_service = DuplicateDetectionService(session)
+                if duplicate_detection_service.is_duplicate_request(
+                    privacy_request,
+                    ConfigProxy(session).privacy_request_duplicate_detection,
+                ):
+                    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)