Include Correlation ID in Idempotence Check (GSI-930) (#15)

* Include correlation ID in notification hash and sort model fields

* Bump version from 1.1.2 -> 1.1.3

Update support files

* Satisfy mypy

Author: Byron Himes

.pre-commit-config.yaml

@@ -48,13 +48,13 @@ repos:
       - id: no-commit-to-branch
         args: [--branch, dev, --branch, int, --branch, main]
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.5.2
+    rev: v0.5.5
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]
       - id: ruff-format
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.10.1
+    rev: v1.11.0
     hooks:
       - id: mypy
         args: [--no-warn-unused-ignores]

.pyproject_generation/pyproject_custom.toml

@@ -1,6 +1,6 @@
 [project]
 name = "ns"
-version = "1.1.2"
+version = "1.1.3"
 description = "The Notification Service (NS) handles notification kafka events."
 dependencies = [
     "typer>=0.12",

README.md

@@ -33,21 +33,21 @@ We recommend using the provided Docker container.
 
 A pre-build version is available at [docker hub](https://hub.docker.com/repository/docker/ghga/notification-service):
 ```bash
-docker pull ghga/notification-service:1.1.2
+docker pull ghga/notification-service:1.1.3
 ```
 
 Or you can build the container yourself from the [`./Dockerfile`](./Dockerfile):
 ```bash
 # Execute in the repo's root dir:
-docker build -t ghga/notification-service:1.1.2 .
+docker build -t ghga/notification-service:1.1.3 .
 ```
 
 For production-ready deployment, we recommend using Kubernetes, however,
 for simple use cases, you could execute the service using docker
 on a single server:
 ```bash
 # The entrypoint is preconfigured:
-docker run -p 8080:8080 ghga/notification-service:1.1.2 --help
+docker run -p 8080:8080 ghga/notification-service:1.1.3 --help
 ```
 
 If you prefer not to use containers, you may install the service from source:

lock/requirements-dev.txt

@@ -21,7 +21,7 @@ classifiers = [
     "Intended Audience :: Developers",
 ]
 name = "ns"
-version = "1.1.2"
+version = "1.1.3"
 description = "The Notification Service (NS) handles notification kafka events."
 dependencies = [
     "typer>=0.12",

lock/requirements.txt

@@ -16,6 +16,7 @@
 """Contains the concrete implementation of a NotifierPort"""
 
 import html
+import json
 import logging
 from contextlib import suppress
 from email.message import EmailMessage
@@ -24,6 +25,7 @@
 from string import Template
 
 from ghga_event_schemas import pydantic_ as event_schemas
+from hexkit.correlation import get_correlation_id
 from pydantic import EmailStr, Field
 from pydantic_settings import BaseSettings
 
@@ -72,8 +74,11 @@ def __init__(
     def _create_notification_record(
         self, *, notification: event_schemas.Notification
     ) -> models.NotificationRecord:
-        """Creates a notification record from a notification event"""
-        hash_sum = sha256(notification.model_dump_json().encode("utf-8")).hexdigest()
+        """Creates a notification record from a notification event and correlation ID."""
+        correlation_id = get_correlation_id()
+        notification_str = json.dumps(notification.model_dump(), sort_keys=True)
+        concatenated = correlation_id + notification_str
+        hash_sum = sha256(concatenated.encode("utf-8")).hexdigest()
         return models.NotificationRecord(hash_sum=hash_sum, sent=False)
 
     async def _has_been_sent(self, *, hash_sum: str) -> bool:

pyproject.toml

@@ -16,10 +16,12 @@
 """Test basic event consumption"""
 
 import html
+import json
 from hashlib import sha256
 from typing import cast
 
 import pytest
+from hexkit.correlation import correlation_id_var
 
 from ns.adapters.outbound.smtp_client import SmtpClient
 from ns.core.models import NotificationRecord
@@ -30,6 +32,8 @@
 
 pytestmark = pytest.mark.asyncio()
 
+TEST_CORRELATION_ID = "6914c8cd-1f18-43da-ac6c-c43cca3f36cc"
+
 sample_notification = {
     "recipient_email": "test@example.com",
     "email_cc": ["test2@test.com", "test3@test.com"],
@@ -40,6 +44,16 @@
 }
 
 
+@pytest.fixture(autouse=True)
+def correlation_id_fixture():
+    """Provides a new correlation ID for each test case."""
+    # we cannot use an async fixture with set_correlation_id(),
+    # because it would run in a different context from the test
+    token = correlation_id_var.set(TEST_CORRELATION_ID)
+    yield
+    correlation_id_var.reset(token)
+
+
 @pytest.mark.parametrize(
     "notification_details",
     [sample_notification],
@@ -61,7 +75,7 @@ async def test_email_construction(
     plaintext_body = msg.get_body(preferencelist="plain")
     assert plaintext_body is not None
 
-    plaintext_content = plaintext_body.get_content()  # type: ignore
+    plaintext_content = plaintext_body.get_content()
     expected_plaintext = (
         "Dear Yolanda Martinez,\n\nWhere are you, where are you, Yolanda?\n"
         + "\nWarm regards,\n\nThe GHGA Team"
@@ -71,7 +85,7 @@ async def test_email_construction(
     html_body = msg.get_body(preferencelist="html")
     assert html_body is not None
 
-    html_content = html_body.get_content()  # type: ignore
+    html_content = html_body.get_content()
     assert html_content is not None
 
     expected_html = (
@@ -149,9 +163,11 @@ async def test_helper_functions(joint_fixture: JointFixture):
     notification = make_notification(sample_notification)
 
     # manually create a NotificationRecord for the notification
+    concatenated = TEST_CORRELATION_ID + json.dumps(
+        notification.model_dump(), sort_keys=True
+    )
     expected_record = NotificationRecord(
-        hash_sum=sha256(notification.model_dump_json().encode("utf-8")).hexdigest(),
-        sent=False,
+        hash_sum=sha256(concatenated.encode("utf-8")).hexdigest(), sent=False
     )
 
     # Now create the record using the notifier's function and compare
@@ -198,6 +214,7 @@ async def test_idempotence_and_transmission(joint_fixture: JointFixture):
     joint_fixture.notifier = cast(Notifier, joint_fixture.notifier)
 
     notification_event = make_notification(sample_notification)
+
     await joint_fixture.kafka.publish_event(
         payload=notification_event.model_dump(),
         type_=joint_fixture.config.notification_event_type,
@@ -209,6 +226,12 @@ async def test_idempotence_and_transmission(joint_fixture: JointFixture):
         notification=notification_event
     )
 
+    # verify the hash is generated with the keys sorted and correlation ID prepended
+    concatenated = TEST_CORRELATION_ID + json.dumps(
+        notification_event.model_dump(), sort_keys=True
+    )
+    assert record.hash_sum == sha256(concatenated.encode("utf-8")).hexdigest()
+
     # the record hasn't been sent, so this should return False
     assert not await joint_fixture.notifier._has_been_sent(hash_sum=record.hash_sum)
 
@@ -267,7 +290,7 @@ async def test_html_escaping(joint_fixture: JointFixture):
     plaintext_body = msg.get_body(preferencelist="plain")
     assert plaintext_body is not None
 
-    plaintext_content = plaintext_body.get_content()  # type: ignore
+    plaintext_content = plaintext_body.get_content()
     expected_plaintext = (
         f"Dear {original_name},\n\n{original_body}\n"
         + "\nWarm regards,\n\nThe GHGA Team"
@@ -277,7 +300,7 @@ async def test_html_escaping(joint_fixture: JointFixture):
     html_body = msg.get_body(preferencelist="html")
     assert html_body is not None
 
-    html_content = html_body.get_content()  # type: ignore
+    html_content = html_body.get_content()
     assert html_content is not None
 
     expected_html = (