fix: make activity result inserts idempotent on retry (CM-1318)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/mailing_list_integration/src/crowdmail/database/crud.py

@@ -219,9 +219,13 @@ async def release_list(list_id: str) -> None:
 
 
 async def batch_insert_activities(records: list[tuple], batch_size=100) -> None:
+    # id is deterministic per (integrationId, shard, git commit) — ON CONFLICT DO
+    # NOTHING makes a retried batch (after a crash or a Kafka-send failure) idempotent
+    # instead of inserting duplicate rows for the same messages.
     sql_query = """
     INSERT INTO integration.results(id, state, data, "tenantId", "integrationId")
     values($1, $2, $3, $4, $5)
+    ON CONFLICT (id) DO NOTHING
     """
     logger.info(f"Saving {len(records)} activities into integration.results")
     for i in range(0, len(records), batch_size):

services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py

@@ -120,7 +120,15 @@ async def _process_single_list(self, mailing_list: MailingList):
                         continue
                     activity_data["segmentId"] = mailing_list.segment_id
 
-                    result_id = str(uuid.uuid1())
+                    # Deterministic (not random) so a retried batch after a crash or a
+                    # Kafka-send failure re-inserts the exact same row instead of a
+                    # duplicate, per commit id which is stable across retries.
+                    result_id = str(
+                        uuid.uuid5(
+                            uuid.NAMESPACE_URL,
+                            f"{mailing_list.integration_id}:{shard}:{git_id}",
+                        )
+                    )
                     data_dict = {"type": IntegrationResultType.ACTIVITY, "data": activity_data}
                     activities_db.append(
                         (