Greptile Summary

This PR implements a two-pronged deduplication mechanism to prevent spurious notifications on service restart: the Kotlin MessageCollector now pre-seeds its in-memory processedMessageIds cache with message IDs from the last 30 days before subscribing to the message flow, and the Python wrapper removes messages from pending_inbound after a successful Kotlin callback so the queue doesn't grow and replay stale entries on restart.

Key changes:

• MessageCollector.startCollecting() pre-seeds processedMessageIds from the DB inside the collection coroutine (ensuring completion before the collect subscription begins), with graceful degradation on failure
• reticulum_wrapper._on_lxmf_delivery() removes the message from pending_inbound after a successful callback; a failed callback preserves the entry for the polling fallback
• MessageDao and ConversationRepository expose a new getReceivedMessageIds(since) bounded query
• All new paths have accompanying unit tests; the Python tests now correctly pre-populate pending_inbound before calling _on_lxmf_delivery, making them independent of internal append side-effects
• One issue found: MessageDao.getReceivedMessageIds filters on the sender's timestamp rather than the local receivedAt column; messages from peers with significantly wrong clocks (>30 days off) could be excluded from pre-seeding and trigger a duplicate notification after restart

Confidence Score: 4/5

• This PR is safe to merge; the deduplication logic is sound and well-tested, with one minor correctness issue in the DAO timestamp filter that only affects edge-case peers with incorrect clocks.
• The core dedup approach is correct: pre-seeding is sequential within the coroutine (no race before subscription), graceful degradation is in place, and the Python queue cleanup is properly scoped inside the try block. The single issue — using timestamp (sender clock) instead of COALESCE(receivedAt, timestamp) (local receipt time) for the 30-day window — is a narrow edge case that only affects messages from peers with clocks set more than 30 days in the past and could cause a duplicate notification on restart for those messages while unread.
• data/src/main/java/com/lxmf/messenger/data/db/dao/MessageDao.kt — the getReceivedMessageIds query timestamp filter

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant App as App Startup
    participant MC as MessageCollector
    participant CR as ConversationRepository
    participant DB as MessageDao (DB)
    participant Flow as reticulumProtocol.observeMessages()
    participant NH as NotificationHelper

    App->>MC: startCollecting()
    MC->>MC: scope.launch { ... }
    note over MC: Pre-seed phase (sequential, before subscribe)
    MC->>CR: getReceivedMessageIds(since = now - 30d)
    CR->>DB: getReceivedMessageIds(identityHash, since)
    DB-->>CR: List<String> (recent msg IDs)
    CR-->>MC: existingIds
    MC->>MC: processedMessageIds.addAll(existingIds)

    note over MC: Subscribe phase
    MC->>Flow: observeMessages().collect { ... }

    Flow-->>MC: replay: receivedMessage (hash = "X")
    alt hash in processedMessageIds (pre-seeded)
        MC->>MC: skip (return@collect) — no notification
    else hash NOT in cache
        MC->>DB: getMessageById(hash)
        alt already in DB
            MC->>MC: add to cache
            alt isRead == false
                MC->>NH: notifyMessageReceived(...)
            end
        else new message
            MC->>DB: saveMessage(...)
            MC->>NH: notifyMessageReceived(...)
        end
    end

_{Last reviewed commit: 2c22108}

@greptileai

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!