feat: fix auto-resend functionality with attachment handling and improve duplicate message cleanup logic

Author: Ivan

meshchatx.rsm

@@ -183,8 +183,10 @@
     RECENT_SAME_CONTENT_SECONDS,
     AutoResendCoordinator,
     cooldown_until,
+    fields_have_attachments,
     fields_with_auto_resend_count,
     next_attempt_count,
+    parse_fields_dict,
     should_skip_for_budget,
 )
 from meshchatx.src.backend.local_message_retention import (
@@ -23244,6 +23246,16 @@ async def resend_failed_messages_for_destination(
                     ):
                         continue
 
+                    fields = parse_fields_dict(failed_message.get("fields"))
+                    allow_attachments = (
+                        ctx.config.allow_auto_resending_failed_messages_with_attachments.get()
+                    )
+                    if not allow_attachments and fields_have_attachments(fields):
+                        print(
+                            "Not resending failed message with attachments, as setting is disabled",
+                        )
+                        continue
+
                     claimed = (
                         ctx.database.messages.try_claim_failed_message_for_auto_resend(
                             message_hash,
@@ -23257,53 +23269,40 @@ async def resend_failed_messages_for_destination(
                     if not claimed:
                         continue
 
-                    # parse fields as json
-                    fields = json.loads(failed_message["fields"] or "{}")
-                    if not isinstance(fields, dict):
-                        fields = {}
-
                     # parse image field
                     image_field = None
-                    if "image" in fields:
+                    if "image" in fields and isinstance(fields.get("image"), dict):
                         image_field = LxmfImageField(
                             fields["image"]["image_type"],
                             base64.b64decode(fields["image"]["image_bytes"]),
                         )
 
                     # parse audio field
                     audio_field = None
-                    if "audio" in fields:
+                    if "audio" in fields and isinstance(fields.get("audio"), dict):
                         audio_field = LxmfAudioField(
                             fields["audio"]["audio_mode"],
                             base64.b64decode(fields["audio"]["audio_bytes"]),
                         )
 
                     # parse file attachments field
                     file_attachments_field = None
-                    if "file_attachments" in fields:
+                    if "file_attachments" in fields and isinstance(
+                        fields.get("file_attachments"),
+                        list,
+                    ):
                         file_attachments = [
                             LxmfFileAttachment(
                                 file_attachment["file_name"],
                                 base64.b64decode(file_attachment["file_bytes"]),
                             )
                             for file_attachment in fields["file_attachments"]
+                            if isinstance(file_attachment, dict)
                         ]
                         file_attachments_field = LxmfFileAttachmentsField(
                             file_attachments,
                         )
 
-                    # don't resend message with attachments if not allowed
-                    if not ctx.config.allow_auto_resending_failed_messages_with_attachments.get():
-                        if (
-                            image_field is not None
-                            or audio_field is not None
-                            or file_attachments_field is not None
-                        ):
-                            print(
-                                "Not resending failed message with attachments, as setting is disabled",
-                            )
-                            continue
-
                     attempt = next_attempt_count(failed_message.get("fields"))
                     ctx.database.messages.set_message_fields_json(
                         message_hash,

meshchatx/meshchat.py

@@ -67,6 +67,20 @@ def lock_for(self, identity_key: str, destination_hash: str) -> asyncio.Lock:
         return lock
 
 
+def fields_have_attachments(fields_raw: Any) -> bool:
+    fields = _parse_fields(fields_raw)
+    if not fields:
+        return False
+    if fields.get("image") or fields.get("audio"):
+        return True
+    files = fields.get("file_attachments")
+    return isinstance(files, list) and len(files) > 0
+
+
+def parse_fields_dict(fields_raw: Any) -> dict:
+    return _parse_fields(fields_raw)
+
+
 def should_skip_for_budget(
     fields_raw: Any, *, max_attempts: int = MAX_AUTO_RESEND_ATTEMPTS
 ) -> bool:

meshchatx/src/backend/auto_resend_guard.py

@@ -852,26 +852,38 @@ def list_duplicate_lxmf_message_hashes_by_content(self) -> list[str]:
         """Hashes of duplicate rows (same peer, direction, and text), excluding the oldest keep.
 
         Empty or whitespace-only content is ignored so attachment-only or blank
-        rows are not collapsed together.
+        rows are not collapsed together. Oldest is by timestamp then id.
         """
-        rows = self.provider.fetchall(
+        groups = self.provider.fetchall(
             """
-            SELECT m.hash AS hash
-            FROM lxmf_messages m
-            INNER JOIN (
-                SELECT peer_hash, is_incoming, content, MIN(id) AS keep_id
-                FROM lxmf_messages
-                WHERE content IS NOT NULL AND TRIM(content) != ''
-                GROUP BY peer_hash, is_incoming, content
-                HAVING COUNT(*) > 1
-            ) d
-              ON m.peer_hash = d.peer_hash
-             AND m.is_incoming = d.is_incoming
-             AND m.content = d.content
-            WHERE m.id != d.keep_id
+            SELECT peer_hash, is_incoming, content
+            FROM lxmf_messages
+            WHERE content IS NOT NULL AND TRIM(content) != ''
+            GROUP BY peer_hash, is_incoming, content
+            HAVING COUNT(*) > 1
             """,
         )
-        return [r["hash"] for r in rows if r.get("hash")]
+        to_delete: list[str] = []
+        for group in groups:
+            rows = self.provider.fetchall(
+                """
+                SELECT hash
+                FROM lxmf_messages
+                WHERE peer_hash = ?
+                  AND is_incoming = ?
+                  AND content = ?
+                ORDER BY
+                    CASE WHEN timestamp IS NULL THEN 1 ELSE 0 END,
+                    timestamp ASC,
+                    id ASC
+                """,
+                (group["peer_hash"], group["is_incoming"], group["content"]),
+            )
+            # Keep the first (oldest); delete the rest.
+            for row in rows[1:]:
+                if row.get("hash"):
+                    to_delete.append(row["hash"])
+        return to_delete
 
     def count_duplicate_lxmf_messages_by_content(self) -> int:
         return len(self.list_duplicate_lxmf_message_hashes_by_content())