diff --git a/src/compose_notifications/_utils/_mixins/notification_record_mixin.py b/src/compose_notifications/_utils/_mixins/notification_record_mixin.py index 871cbb85..ce928837 100644 --- a/src/compose_notifications/_utils/_mixins/notification_record_mixin.py +++ b/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.""" diff --git a/src/compose_notifications/_utils/notifications_maker.py b/src/compose_notifications/_utils/notifications_maker.py index 802ba895..0f67825f 100644 --- a/src/compose_notifications/_utils/notifications_maker.py +++ b/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: