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

meshchatx/meshchat.py

@@ -183,8 +183,10 @@ from meshchatx.src.backend.auto_resend_guard import (
     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 @@ class ReticulumMeshChat:
                     ):
                         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,14 +23269,9 @@ class ReticulumMeshChat:
                     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"]),
@@ -23272,7 +23279,7 @@ class ReticulumMeshChat:
 
                     # 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"]),
@@ -23280,30 +23287,22 @@ class ReticulumMeshChat:
 
                     # 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/src/backend/auto_resend_guard.py

@@ -67,6 +67,20 @@ class AutoResendCoordinator:
         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/database/messages.py

@@ -852,26 +852,38 @@ class MessageDAO:
         """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())

tests/backend/test_auto_resend_regression.py

@@ -0,0 +1,338 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Regression tests for auto-resend flooding guards and duplicate cleanup."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+import types
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from meshchatx.meshchat import ReticulumMeshChat
+from meshchatx.src.backend import auto_resend_guard as guard
+from meshchatx.src.backend.database import Database
+from meshchatx.src.backend.database.provider import DatabaseProvider
+from meshchatx.src.backend.database.schema import DatabaseSchema
+
+
+@pytest.fixture
+def db(tmp_path):
+    path = str(tmp_path / "resend_reg.db")
+    provider = DatabaseProvider(path)
+    DatabaseSchema(provider).initialize()
+    database = Database(path)
+    yield database
+    database.close_all()
+    provider.close_all()
+
+
+def _insert_failed(db, *, msg_hash, peer, content, fields="{}", ts=None, next_at=None):
+    now = ts if ts is not None else time.time()
+    db.messages.upsert_lxmf_message(
+        {
+            "hash": msg_hash,
+            "source_hash": peer,
+            "destination_hash": peer,
+            "peer_hash": peer,
+            "state": "failed",
+            "progress": 0,
+            "is_incoming": 0,
+            "method": "opportunistic",
+            "delivery_attempts": 0,
+            "next_delivery_attempt_at": next_at,
+            "title": "",
+            "content": content,
+            "fields": fields,
+            "rssi": None,
+            "snr": None,
+            "quality": None,
+            "is_spam": 0,
+            "reply_to_hash": None,
+            "attachments_stripped": 0,
+            "timestamp": now,
+        },
+    )
+
+
+def _insert_outbound(db, *, msg_hash, peer, content, state="delivered", ts=None):
+    now = ts if ts is not None else time.time()
+    db.messages.upsert_lxmf_message(
+        {
+            "hash": msg_hash,
+            "source_hash": peer,
+            "destination_hash": peer,
+            "peer_hash": peer,
+            "state": state,
+            "progress": 1,
+            "is_incoming": 0,
+            "method": "direct",
+            "delivery_attempts": 1,
+            "next_delivery_attempt_at": None,
+            "title": "",
+            "content": content,
+            "fields": "{}",
+            "rssi": None,
+            "snr": None,
+            "quality": None,
+            "is_spam": 0,
+            "reply_to_hash": None,
+            "attachments_stripped": 0,
+            "timestamp": now,
+        },
+    )
+
+
+def _bind_resend_app(db):
+    app = MagicMock()
+    app._auto_resend_coordinator = guard.AutoResendCoordinator()
+    app.websocket_broadcast = AsyncMock()
+    app.send_message = AsyncMock()
+    app.resend_failed_messages_for_destination = types.MethodType(
+        ReticulumMeshChat.resend_failed_messages_for_destination,
+        app,
+    )
+    ctx = MagicMock()
+    ctx.identity.hash.hex.return_value = "aa" * 16
+    ctx.database = db
+    ctx.config.allow_auto_resending_failed_messages_with_attachments.get.return_value = (
+        True
+    )
+    return app, ctx
+
+
+def test_fields_have_attachments_and_invalid_json():
+    assert not guard.fields_have_attachments("{}")
+    assert not guard.fields_have_attachments("not-json")
+    assert guard.fields_have_attachments('{"image":{"image_type":"png"}}')
+    assert guard.fields_have_attachments('{"file_attachments":[{"file_name":"a"}]}')
+    assert guard.parse_fields_dict("not-json") == {}
+    assert guard.should_skip_for_budget(
+        '{"_mcx_auto_resend_count": 3}',
+        max_attempts=3,
+    )
+
+
+@pytest.mark.asyncio
+async def test_resend_skips_when_budget_exhausted(db):
+    peer = "b" * 32
+    msg = "c" * 32
+    _insert_failed(
+        db,
+        msg_hash=msg,
+        peer=peer,
+        content="help",
+        fields=json.dumps({"_mcx_auto_resend_count": 3}),
+    )
+    app, ctx = _bind_resend_app(db)
+    await app.resend_failed_messages_for_destination(peer, context=ctx)
+    app.send_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_resend_skips_when_recent_same_content_exists(db):
+    peer = "d" * 32
+    now = time.time()
+    _insert_outbound(
+        db,
+        msg_hash="1" * 32,
+        peer=peer,
+        content="help text",
+        ts=now - 30,
+    )
+    _insert_failed(
+        db,
+        msg_hash="2" * 32,
+        peer=peer,
+        content="help text",
+        ts=now - 5,
+    )
+    app, ctx = _bind_resend_app(db)
+    await app.resend_failed_messages_for_destination(peer, context=ctx)
+    app.send_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_resend_skips_attachments_before_claim(db):
+    peer = "e" * 32
+    msg = "f" * 32
+    fields = json.dumps(
+        {
+            "image": {
+                "image_type": "png",
+                "image_bytes": "aa",
+            },
+        },
+    )
+    _insert_failed(db, msg_hash=msg, peer=peer, content="pic", fields=fields)
+    app, ctx = _bind_resend_app(db)
+    ctx.config.allow_auto_resending_failed_messages_with_attachments.get.return_value = (
+        False
+    )
+    await app.resend_failed_messages_for_destination(peer, context=ctx)
+    app.send_message.assert_not_called()
+    row = db.provider.fetchone(
+        "SELECT next_delivery_attempt_at FROM lxmf_messages WHERE hash = ?",
+        (msg,),
+    )
+    # Must not burn cooldown when attachments are disallowed.
+    assert row["next_delivery_attempt_at"] is None
+
+
+@pytest.mark.asyncio
+async def test_resend_claims_once_and_deletes_old_on_success(db):
+    peer = "1" * 32
+    old_hash = "2" * 32
+    new_hash_bytes = bytes.fromhex("3" * 32)
+    _insert_failed(db, msg_hash=old_hash, peer=peer, content="retry me")
+    app, ctx = _bind_resend_app(db)
+
+    async def send_and_materialize(*args, **kwargs):
+        new_msg = MagicMock()
+        new_msg.hash = new_hash_bytes
+        _insert_outbound(
+            db,
+            msg_hash=new_hash_bytes.hex(),
+            peer=peer,
+            content="retry me",
+            state="outbound",
+        )
+        return new_msg
+
+    app.send_message.side_effect = send_and_materialize
+
+    await app.resend_failed_messages_for_destination(peer, context=ctx)
+    assert app.send_message.await_count == 1
+    assert (
+        db.provider.fetchone(
+            "SELECT 1 FROM lxmf_messages WHERE hash = ?",
+            (old_hash,),
+        )
+        is None
+    )
+    new_row = db.provider.fetchone(
+        "SELECT fields FROM lxmf_messages WHERE hash = ?",
+        (new_hash_bytes.hex(),),
+    )
+    assert guard.read_auto_resend_count(new_row["fields"]) == 1
+    app.websocket_broadcast.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_resend_concurrent_calls_send_only_once(db):
+    peer = "4" * 32
+    old_hash = "5" * 32
+    _insert_failed(db, msg_hash=old_hash, peer=peer, content="race")
+    app, ctx = _bind_resend_app(db)
+
+    async def slow_send(*args, **kwargs):
+        await asyncio.sleep(0.05)
+        m = MagicMock()
+        m.hash = bytes.fromhex("6" * 32)
+        _insert_outbound(
+            db,
+            msg_hash=m.hash.hex(),
+            peer=peer,
+            content="race",
+            state="outbound",
+        )
+        return m
+
+    app.send_message.side_effect = slow_send
+    await asyncio.gather(
+        app.resend_failed_messages_for_destination(peer, context=ctx),
+        app.resend_failed_messages_for_destination(peer, context=ctx),
+    )
+    assert app.send_message.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_resend_does_not_delete_old_when_send_returns_none(db):
+    peer = "7" * 32
+    old_hash = "8" * 32
+    _insert_failed(db, msg_hash=old_hash, peer=peer, content="keep")
+    app, ctx = _bind_resend_app(db)
+    app.send_message.return_value = None
+    await app.resend_failed_messages_for_destination(peer, context=ctx)
+    assert db.provider.fetchone(
+        "SELECT state FROM lxmf_messages WHERE hash = ?",
+        (old_hash,),
+    )["state"] == "failed"
+    # Cooldown claimed, attempt counted.
+    row = db.provider.fetchone(
+        "SELECT fields, next_delivery_attempt_at FROM lxmf_messages WHERE hash = ?",
+        (old_hash,),
+    )
+    assert guard.read_auto_resend_count(row["fields"]) == 1
+    assert row["next_delivery_attempt_at"] is not None
+
+
+def test_duplicate_cleanup_keeps_oldest_by_timestamp(db):
+    peer = "9" * 32
+    base = time.time()
+    # Later insert id can still be oldest by timestamp.
+    _insert_outbound(db, msg_hash="a" * 32, peer=peer, content="dup", ts=base - 5)
+    _insert_outbound(db, msg_hash="b" * 32, peer=peer, content="dup", ts=base - 50)
+    _insert_outbound(db, msg_hash="c" * 32, peer=peer, content="dup", ts=base - 1)
+    deleted = db.messages.delete_duplicate_lxmf_messages_by_content()
+    assert deleted == 2
+    left = db.provider.fetchone(
+        "SELECT hash FROM lxmf_messages WHERE content = 'dup'",
+    )
+    assert left["hash"] == "b" * 32
+
+
+def test_source_keeps_auto_resend_and_duplicate_guards():
+    from pathlib import Path
+
+    root = Path(__file__).resolve().parents[2]
+    meshchat = (root / "meshchatx" / "meshchat.py").read_text(encoding="utf-8")
+    guard_src = (
+        root / "meshchatx" / "src" / "backend" / "auto_resend_guard.py"
+    ).read_text(encoding="utf-8")
+    messages = (
+        root / "meshchatx" / "src" / "backend" / "database" / "messages.py"
+    ).read_text(encoding="utf-8")
+    settings = (
+        root
+        / "meshchatx"
+        / "src"
+        / "frontend"
+        / "components"
+        / "settings"
+        / "SettingsPage.vue"
+    ).read_text(encoding="utf-8")
+
+    assert "AutoResendCoordinator" in meshchat
+    assert "try_claim_failed_message_for_auto_resend" in meshchat
+    assert "fields_have_attachments" in meshchat
+    assert "has_path(destination_hash)" in meshchat
+    assert "MAX_AUTO_RESEND_ATTEMPTS" in guard_src
+    assert "delete_duplicate_lxmf_messages_by_content" in messages
+    assert "clearDuplicateMessages" in settings
+    assert "/api/v1/maintenance/messages/duplicates" in meshchat
+
+
+def test_claim_reopens_after_cooldown_expires(db):
+    peer = "0" * 32
+    msg = "a" * 32
+    now = 1_000_000.0
+    _insert_failed(db, msg_hash=msg, peer=peer, content="x", next_at=now - 1)
+    assert db.messages.try_claim_failed_message_for_auto_resend(
+        msg,
+        cooldown_until=now + 120,
+        now=now,
+    )
+    assert not db.messages.try_claim_failed_message_for_auto_resend(
+        msg,
+        cooldown_until=now + 240,
+        now=now + 10,
+    )
+    assert db.messages.try_claim_failed_message_for_auto_resend(
+        msg,
+        cooldown_until=now + 360,
+        now=now + 121,
+    )

tests/backend/test_duplicate_messages_cleanup.py

@@ -65,6 +65,24 @@ def test_delete_duplicate_messages_by_content_keeps_oldest(tmp_path):
     provider.close_all()
 
 
+def test_delete_duplicate_prefers_earlier_timestamp_over_lower_id(tmp_path):
+    path = str(tmp_path / "d3.db")
+    provider = DatabaseProvider(path)
+    DatabaseSchema(provider).initialize()
+    db = Database(path)
+    peer = "d" * 32
+    base = time.time()
+    # Insert newer timestamp first (lower id), then older timestamp.
+    _msg(db, msg_hash="1" * 32, peer=peer, content="z", ts=base - 1)
+    _msg(db, msg_hash="2" * 32, peer=peer, content="z", ts=base - 100)
+    deleted = db.messages.delete_duplicate_lxmf_messages_by_content()
+    assert deleted == 1
+    left = db.provider.fetchone("SELECT hash FROM lxmf_messages")
+    assert left["hash"] == "2" * 32
+    db.close_all()
+    provider.close_all()
+
+
 def test_duplicates_are_scoped_by_peer_and_direction(tmp_path):
     path = str(tmp_path / "d2.db")
     provider = DatabaseProvider(path)

tests/frontend/SettingsPage.config-persistence.test.js

@@ -611,6 +611,14 @@ describe("SettingsPage — maintenance, exports, telemetry trust, RNS reload", (
         expect(api.delete).toHaveBeenCalledWith("/api/v1/maintenance/messages/duplicates");
     });
 
+    it("maintenance UI exposes duplicate cleanup and age purge", async () => {
+        const w = await mountSettingsPage(api);
+        const html = w.html();
+        expect(html).toContain("maintenance.clear_duplicates");
+        expect(html).toContain("maintenance.purge_old_title");
+        expect(html).toContain("maintenance.export_old_archive");
+    });
+
     it("clearAnnounces DELETEs announces", async () => {
         const w = await mountSettingsPage(api);
         await w.vm.clearAnnounces();