fix(compose_notifications): use ON CONFLICT DO NOTHING to prevent UnqueViolation

src/compose_notifications/_utils/_mixins/notification_record_mixin.py

@@ -34,20 +34,42 @@ def get_message_group_count(self, user_id: int, message_type: str) -> int:
             ).fetchone()
             return result[0] if result[0] is not None else 0
 
-    def batch_insert_notifications(self, records: list[dict]) -> None:
-        """Batch insert notification records. Each dict has keys matching notif_by_user columns."""
+    def batch_insert_notifications(self, records: list[dict]) -> int:
+        """Batch insert notification records. Each dict has keys matching notif_by_user columns.
+
+        Uses INSERT … ON CONFLICT DO NOTHING on the partial unique index
+        notif_by_user_unique_unsent_idx to silently skip duplicates that may
+        arise from YMQ redelivery or concurrent invocations processing the
+        same change_log_id.
+
+        Returns the number of rows actually inserted (may be less than len(records)
+        if duplicates were skipped).
+        """
         if not records:
-            return
+            return 0
 
         with self.connect() as conn:
-            columns = ', '.join(records[0].keys())
-            placeholders = ', '.join(f':{k}' for k in records[0].keys())
+            columns = list(records[0].keys())
+            col_list = ', '.join(columns)
+
+            # Build multi-row VALUES with per-row parameter suffixes to avoid collisions
+            all_placeholders: list[str] = []
+            all_params: dict[str, object] = {}
+            for i, rec in enumerate(records):
+                row_placeholders = ', '.join(f':{k}_{i}' for k in columns)
+                all_placeholders.append(f'({row_placeholders})')
+                for k in columns:
+                    all_params[f'{k}_{i}'] = rec[k]
+
             stmt = sqlalchemy.text(f"""
-                INSERT INTO notif_by_user ({columns})
-                VALUES ({placeholders})
+                INSERT INTO notif_by_user ({col_list})
+                VALUES {', '.join(all_placeholders)}
+                ON CONFLICT (change_log_id, user_id, message_type, (COALESCE(messenger, 'telegram')))
+                WHERE completed IS NULL AND cancelled IS NULL
+                DO NOTHING
             """)
-            for rec in records:
-                conn.execute(stmt, rec)
+            result = conn.execute(stmt, all_params)
+            return result.rowcount
 
     def resolve_messengers(self, user_ids: list[int]) -> list[tuple]:
         """Batch-resolve messengers for users from user_identity_map."""

src/compose_notifications/_utils/notifications_maker.py

@@ -249,24 +249,15 @@ def _resolve_messengers_batch(self) -> None:
             self._messenger_map[uid] = list(messengers)
 
     def flush_batch(self) -> None:
-        """Flush the batch buffer to the database"""
+        """Flush the batch buffer to the database.
+
+        Uses INSERT … ON CONFLICT DO NOTHING so duplicate records
+        (e.g. from YMQ redelivery) are silently skipped instead of
+        raising UniqueViolation.
+        """
         if not self._batch_buffer:
             return
 
-        # DIAG: check if any records in this batch would create duplicates
-        change_log_id = self.new_record.change_log_id
-        for record in self._batch_buffer:
-            existing_count = self.db.check_notification_duplicate(
-                change_log_id, record.user_id, record.message_type, record.messenger
-            )
-            if existing_count and existing_count > 0:
-                logging.warning(
-                    f'DOUBLING_DIAG: flush_batch would create duplicate! '
-                    f'change_log_id={change_log_id} user_id={record.user_id} '
-                    f'message_type={record.message_type} messenger={record.messenger} '
-                    f'existing_count={existing_count}'
-                )
-
         # Convert NotificationRecord dataclass instances to dicts with
         # column names matching the notif_by_user table.
         records = [
@@ -283,8 +274,18 @@ def flush_batch(self) -> None:
             }
             for r in self._batch_buffer
         ]
-        self.db.batch_insert_notifications(records)
-        logging.debug(f'Flushed {len(self._batch_buffer)} records to notif_by_user table')
+
+        inserted = self.db.batch_insert_notifications(records)
+        skipped = len(records) - inserted
+
+        if skipped > 0:
+            logging.warning(
+                f'flush_batch: {inserted} inserted, {skipped} duplicates skipped '
+                f'(change_log_id={self.new_record.change_log_id})'
+            )
+        else:
+            logging.debug(f'Flushed {inserted} records to notif_by_user table')
+
         self._batch_buffer.clear()
 
     def record_notification_statistics(self) -> None: