feat: 315764 Add push action when RDP is created, create async flow for create & push rdp

Author: roma-valor

src/country_workspace/config/__init__.py

@@ -179,6 +179,7 @@ class Group(Enum):
     "MAILJET_SECRET_KEY": (str, "", "", False, "Mailjet API secret key"),
     "DEDUP_API_URL": (str, "", "", False, "Dedup Engine API base URL"),
     "DEDUP_API_TOKEN": (str, "", "", False, "Dedup Engine API token"),
+    "APP_BASE_URL": (str, "", "", False, "Base URL of this application (used to build absolute callback URLs)"),
     "EMAIL_BACKEND": (
         str,
         "anymail.backends.mailjet.EmailBackend",

src/country_workspace/config/fragments/constance.py

@@ -1,5 +1,5 @@
 from .app import AURORA_API_TOKEN, AURORA_API_URL, HOPE_API_TOKEN, HOPE_API_URL, NEW_USER_DEFAULT_GROUP
-from .dedup import DEDUP_API_URL, DEDUP_API_TOKEN
+from .dedup import DEDUP_API_URL, DEDUP_API_TOKEN, APP_BASE_URL
 from .kobo import KOBO_API_TOKEN, KOBO_KF_URL, KOBO_MASTER_API_TOKEN, KOBO_PROJECT_VIEW_ID
 from .mail import MAILJET_API_KEY, MAILJET_SECRET_KEY
 
@@ -68,6 +68,7 @@
     "MAILJET_SECRET_KEY": (MAILJET_SECRET_KEY, "Mailjet secret key", "write_only_text_input"),
     "DEDUP_API_URL": (DEDUP_API_URL, "Dedup Engine server address", str),
     "DEDUP_API_TOKEN": (DEDUP_API_TOKEN, "Dedup Engine API access token", "write_only_text_input"),
+    "APP_BASE_URL": (APP_BASE_URL, "Current server address", str),
     "CHUNK_SIZE_FOR_VALIDATION_TASK": (500, "Number of records to process per chunk in validation tasks", int),
     "PICTURE_IMPORT_MAX_ZIP_UPLOAD_MB": (
         20,
@@ -173,6 +174,7 @@
         "NEW_USER_IS_STAFF",
         "NEW_USER_DEFAULT_GROUP",
         "CONCURRENCY_GUARD",
+        "APP_BASE_URL",
     ),
 }
 

src/country_workspace/config/fragments/dedup.py

@@ -2,3 +2,4 @@
 
 DEDUP_API_URL = env("DEDUP_API_URL")
 DEDUP_API_TOKEN = env("DEDUP_API_TOKEN")
+APP_BASE_URL = env("APP_BASE_URL")

src/country_workspace/config/urls.py

@@ -5,6 +5,7 @@
 from django.contrib import admin
 from django.urls import path
 
+from country_workspace.contrib.hope.views import DeduplicationCallbackView
 from country_workspace.workspaces.sites import workspace
 
 urlpatterns = [
@@ -16,6 +17,11 @@
     path(r"sentry_debug/", lambda _: 1 / 0),
     path("select2/", include(django_select2.urls)),
     path(r"__debug__/", include(debug_toolbar.urls)),
+    path(
+        "api/dedup/callback/<str:signed_token>/",
+        DeduplicationCallbackView.as_view(),
+        name="dedup_callback",
+    ),
 ]
 
 if "django_browser_reload" in settings.INSTALLED_APPS:  # pragma: no cover

src/country_workspace/contrib/dedup_engine/client.py

@@ -69,14 +69,17 @@ def _request[T](self, operation: str, fn: Callable[[], T]) -> T:
         except RequestException as exc:
             raise self._err(operation, exc, getattr(exc, "response", None), error_cls=RemoteUnavailableError) from exc
 
-    def create_deduplication_set(self) -> response.CreatedDeduplicationSet:
+    def create_deduplication_set(self, notification_url: str | None = None) -> response.CreatedDeduplicationSet:
         collection = resource.DeduplicationSetCollection(
             self.session,
             self.api_root.deduplication_sets,
         )
         payload: request.CreateDeduplicationSet = {"reference_pk": self.group_reference_id}
         if self.deduplication_set_id:
             payload["id"] = self.deduplication_set_id
+        if notification_url:
+            payload["notification_url"] = notification_url
+            payload["notify"] = True
 
         result = self._request(
             "create_deduplication_set",

src/country_workspace/contrib/hope/forms.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from django import forms
 
 from country_workspace.workspaces.admin.cleaners.base import BaseActionForm
@@ -7,3 +9,30 @@ class CreateRDPForm(BaseActionForm):
     batch_name = forms.CharField(
         required=False, help_text="Label for this RDP creation. Defaults is the current date and time."
     )
+    push_to_hope = forms.BooleanField(
+        required=False,
+        label="Push to HOPE after creation",
+        help_text="Automatically push beneficiaries to HOPE when RDP creation succeeds.",
+    )
+
+    def __init__(self, *args: Any, show_push_option: bool = False, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        if not show_push_option:
+            self.fields.pop("push_to_hope")
+
+
+class CreateRDPushThresholdForm(BaseActionForm):
+    batch_name = forms.CharField(widget=forms.HiddenInput, required=False)
+    push_to_hope = forms.CharField(widget=forms.HiddenInput)
+    max_dedup_findings_percent = forms.IntegerField(
+        min_value=0,
+        max_value=100,
+        initial=0,
+        required=False,
+        label="Max duplicate findings (%)",
+        help_text=(
+            "Maximum share of individuals allowed to have duplicate matches before push is blocked. "
+            "Only applies when biometric deduplication is enabled. "
+            "0 means any duplicate finding will block the push."
+        ),
+    )

src/country_workspace/contrib/hope/push/__init__.py

@@ -3,7 +3,10 @@
     cancel_existing_rdp_core,
     claim_rdp_deduplication,
     claim_rdp_push,
+    create_and_push_rdp_core,
+    create_rdp_and_start_dedup_core,
     create_rdp_core,
+    dedup_callback_handle,
     dedup_existing_rdp_core,
     push_existing_rdp_core,
 )
@@ -18,7 +21,10 @@
     "cancel_existing_rdp_core",
     "claim_rdp_deduplication",
     "claim_rdp_push",
+    "create_and_push_rdp_core",
+    "create_rdp_and_start_dedup_core",
     "create_rdp_core",
+    "dedup_callback_handle",
     "dedup_existing_rdp_core",
     "get_program_dedup_settings_policy",
     "get_rdp_policy",

src/country_workspace/contrib/hope/push/config.py

@@ -23,6 +23,7 @@ class CreateRdpConfig(SelectionConfig):
     country_office_id: ReadOnly[int]
     program_id: ReadOnly[int]
     pushed_by_id: ReadOnly[int]
+    max_dedup_findings_percent: NotRequired[int]
 
 
 class PushWorkflowConfig(SelectionConfig):

src/country_workspace/contrib/hope/push/orchestration.py

@@ -1,11 +1,21 @@
+import logging
 from collections.abc import Callable, Iterator
 from functools import partial
 from typing import Any
+from constance import config
 from uuid import UUID, uuid4
 
+from django.core import signing
 from django.db import IntegrityError, transaction
-
-from country_workspace.contrib.dedup_engine import REJECTABLE_DEDUPLICATION_SET_STATES, make_dedup_client
+from django.urls import reverse
+from strategy_field.utils import fqn
+
+from country_workspace.contrib.dedup_engine import (
+    REJECTABLE_DEDUPLICATION_SET_STATES,
+    DedupResponseStatus,
+    DeduplicationSetState,
+    make_dedup_client,
+)
 from country_workspace.contrib.hope.exceptions import HopePushError
 from country_workspace.exceptions import RemoteError, RemoteUnavailableError
 from country_workspace.models import AsyncJob, Rdp
@@ -20,6 +30,7 @@
     qs_households,
     qs_individuals_by_household_pks,
     qs_individuals_by_pks,
+    qs_individuals_for_rdp,
     rdp_for_dedup,
     rdp_for_push,
     release_rdp_dedup_settings_lock,
@@ -28,6 +39,11 @@
     workflow_config_for_rdp,
 )
 
+_DEDUP_CALLBACK_SIGN_KEY = "dedup_callback"
+_DEDUP_CALLBACK_MAX_AGE = 60 * 60 * 96  # 96 hours
+
+logger = logging.getLogger(__name__)
+
 
 def _require_policy_check(check: Callable[[], ActionCheck]) -> None:
     try:
@@ -78,6 +94,210 @@ def create_rdp_core(job: AsyncJob) -> dict[str, Any]:
     return {"rdp_id": rdp.id, "rdp_str": str(rdp)}
 
 
+def _build_dedup_callback_url(rdp_id: int, job_id: int) -> str:
+    """Build an absolute, signed callback URL for the dedup engine to call when dedup finishes."""
+    token = signing.dumps(
+        {"rdp_id": rdp_id, "job_id": job_id},
+        key=_DEDUP_CALLBACK_SIGN_KEY,
+    )
+    path = reverse("dedup_callback", kwargs={"signed_token": token})
+    base = config.APP_BASE_URL.rstrip("/")
+    return f"{base}{path}"
+
+
+def create_rdp_and_start_dedup_core(job: AsyncJob) -> dict[str, Any]:
+    """Create an RDP, upload images to the dedup engine, and start deduplication.
+
+    The dedup engine will call back the notification URL when deduplication finishes.
+    The push to HOPE is deferred until the callback is received and the findings
+    threshold is evaluated.
+    """
+    create_result = create_rdp_core(job)
+    rdp_id = create_result["rdp_id"]
+
+    with transaction.atomic():
+        _check, rdp = claim_rdp_deduplication(rdp_id)
+        if rdp is None:
+            raise HopePushError({"errors": ["RDP: could not claim deduplication set."], "rdp_id": rdp_id})
+
+    job.refresh_from_db()
+
+    callback_url = _build_dedup_callback_url(rdp_id=rdp_id, job_id=job.pk)
+    processor = DedupProcessor(rdp)
+    processor.run(notification_url=callback_url)
+
+    if processor.has_errors:
+        raise HopePushError({**processor.total, "rdp_id": rdp_id})
+
+    with transaction.atomic():
+        locked = lock_rdp_for_update(pk=rdp_id)
+        locked.status = Rdp.PushStatus.DEDUP_PENDING
+        locked.save(update_fields=["status"])
+
+    return {**create_result, "dedup_pending": True}
+
+
+def create_and_push_rdp_core(job: AsyncJob) -> dict[str, Any]:
+    """Create an RDP and start deduplication; the push to HOPE is deferred.
+
+    Push to HOPE is only available for programs with biometric deduplication
+    enabled. Deduplication is started first and the push is deferred until the
+    dedup engine calls back the notification URL and the findings threshold is
+    evaluated.
+    """
+    rdp_program = job.program
+    if not (rdp_program and rdp_program.biometric_deduplication_enabled):
+        raise HopePushError({"errors": ["RDP: push to HOPE requires biometric deduplication for this program."]})
+    return create_rdp_and_start_dedup_core(job)
+
+
+def _lock_and_fail_rdp(rdp_id: int, *, reason: str) -> None:
+    with transaction.atomic():
+        locked = lock_rdp_for_update(pk=rdp_id)
+        if locked.status == Rdp.PushStatus.DEDUP_PENDING:
+            set_rdp_push_status(rdp=locked, status=Rdp.PushStatus.FAILURE, hope_rdi_id="N/A")
+            logger.warning(
+                "dedup_callback_handle: rdp_id=%s marked FAILURE (%s)",
+                rdp_id,
+                reason,
+            )
+        else:
+            logger.info(
+                "dedup_callback_handle: rdp_id=%s skip FAILURE (%s); status=%s",
+                rdp_id,
+                reason,
+                locked.status,
+            )
+
+
+def _handle_deduplicated(rdp: Rdp, rdp_id: int, findings_count: int) -> None:
+    origin_job = AsyncJob.objects.filter(rdp_id=rdp_id).order_by("-id").first()
+    max_findings_percent: int = origin_job.config.get("max_dedup_findings_percent", 0) if origin_job else 0
+
+    total_individuals = qs_individuals_for_rdp(rdp=rdp).count()
+    findings_rate = findings_count / total_individuals * 100 if total_individuals > 0 else float(findings_count)
+
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s DEDUPLICATED findings_count=%s total_individuals=%s "
+        "findings_rate=%.2f%% max_dedup_findings_percent=%s",
+        rdp_id,
+        findings_count,
+        total_individuals,
+        findings_rate,
+        max_findings_percent,
+    )
+
+    if findings_rate > max_findings_percent:
+        _lock_and_fail_rdp(
+            rdp_id,
+            reason=f"findings_rate {findings_rate:.2f}% exceeds threshold {max_findings_percent}%",
+        )
+        return
+
+    with transaction.atomic():
+        locked = lock_rdp_for_update(pk=rdp_id)
+        if locked.status != Rdp.PushStatus.DEDUP_PENDING:
+            logger.warning(
+                "dedup_callback_handle: rdp_id=%s skip push queue; status changed to %s",
+                rdp_id,
+                locked.status,
+            )
+            return
+        locked.status = Rdp.PushStatus.PENDING
+        locked.save(update_fields=["status"])
+
+    push_job = AsyncJob.objects.create(
+        description=f"Push RDP {rdp_id} to HOPE (post-dedup)",
+        type=AsyncJob.JobType.TASK,
+        owner=rdp.pushed_by,
+        action=fqn(push_existing_rdp_core),
+        program=rdp.program,
+        rdp=rdp,
+        config={"rdp_id": rdp_id},
+    )
+    push_job.queue()
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s status DEDUP_PENDING→PENDING; queued push job_id=%s",
+        rdp_id,
+        push_job.pk,
+    )
+
+
+def dedup_callback_handle(rdp_id: int) -> None:
+    """Handle a dedup engine callback for the given RDP.
+
+    Fetches the current deduplication set state from the engine.  Only acts on
+    terminal states (``DEDUPLICATED``, ``DEDUPLICATION_FAILED``,
+    ``ENCODING_FAILED``).  For intermediate states the function returns
+    immediately so the engine can call again after the next state transition.
+
+    When dedup is successful and findings are within the configured threshold,
+    a ``push_existing_rdp_core`` async job is queued.  Otherwise the RDP is
+    marked as ``FAILURE``.
+    """
+    try:
+        rdp = rdp_for_push(pk=rdp_id)
+    except Rdp.DoesNotExist:
+        logger.warning("dedup_callback_handle: rdp_id=%s not found", rdp_id)
+        return
+
+    if rdp.status != Rdp.PushStatus.DEDUP_PENDING:
+        logger.info(
+            "dedup_callback_handle: rdp_id=%s skip; status=%s (expected DEDUP_PENDING)",
+            rdp_id,
+            rdp.status,
+        )
+        return
+
+    if not rdp.deduplication_set_id:
+        logger.warning("dedup_callback_handle: rdp_id=%s missing deduplication_set_id", rdp_id)
+        _lock_and_fail_rdp(rdp_id, reason="missing deduplication_set_id")
+        return
+
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s deduplication_set_id=%s; fetching remote status",
+        rdp_id,
+        rdp.deduplication_set_id,
+    )
+
+    status = get_rdp_policy(rdp).deduplication_status(rdp)
+    if status is None:
+        logger.warning(
+            "dedup_callback_handle: rdp_id=%s dedup engine returned no status; leaving DEDUP_PENDING",
+            rdp_id,
+        )
+        return
+
+    if status.response_status != DedupResponseStatus.OK:
+        logger.warning(
+            "dedup_callback_handle: rdp_id=%s dedup engine status unavailable (%s); leaving DEDUP_PENDING",
+            rdp_id,
+            status.response_status,
+        )
+        return
+
+    dedup_state = status.deduplication_set_status
+    terminal_failure_states = {DeduplicationSetState.DEDUPLICATION_FAILED, DeduplicationSetState.ENCODING_FAILED}
+
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s remote_state=%s findings_count=%s",
+        rdp_id,
+        dedup_state,
+        status.findings_count,
+    )
+
+    if dedup_state in terminal_failure_states:
+        _lock_and_fail_rdp(rdp_id, reason=f"dedup engine state {dedup_state}")
+    elif dedup_state == DeduplicationSetState.DEDUPLICATED:
+        _handle_deduplicated(rdp, rdp_id, status.findings_count)
+    else:
+        logger.info(
+            "dedup_callback_handle: rdp_id=%s intermediate state %s; waiting for next callback",
+            rdp_id,
+            dedup_state,
+        )
+
+
 def claim_rdp_deduplication(rdp_id: int) -> tuple[ActionCheck, Rdp | None]:
     rdp = rdp_for_dedup(pk=rdp_id)
     policy = get_rdp_policy(rdp)