Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions gateway/platforms/email.py
Original file line number Diff line number Diff line change
Expand Up @@ -213,13 +213,34 @@ def __init__(self, config: PlatformConfig):

# Track message IDs we've already processed to avoid duplicates
self._seen_uids: set = set()
self._seen_uids_max: int = 2000 # cap to prevent unbounded memory growth
self._poll_task: Optional[asyncio.Task] = None

# Map chat_id (sender email) -> last subject + message-id for threading
self._thread_context: Dict[str, Dict[str, str]] = {}

logger.info("[Email] Adapter initialized for %s", self._address)

def _trim_seen_uids(self) -> None:
"""Keep only the most recent UIDs to prevent unbounded memory growth.

IMAP UIDs are monotonically increasing integers. When the set grows
beyond the cap, we keep only the highest half — old UIDs are safe to
drop because new messages always have higher UIDs and IMAP's UNSEEN
flag prevents re-delivery regardless.
"""
if len(self._seen_uids) <= self._seen_uids_max:
return
try:
# UIDs are bytes like b'1234' — sort numerically and keep top half
sorted_uids = sorted(self._seen_uids, key=lambda u: int(u))
keep = self._seen_uids_max // 2
self._seen_uids = set(sorted_uids[-keep:])
logger.debug("[Email] Trimmed seen UIDs to %d entries", len(self._seen_uids))
except (ValueError, TypeError):
# Fallback: just clear old entries if sort fails
self._seen_uids = set(list(self._seen_uids)[-self._seen_uids_max // 2:])

async def connect(self) -> bool:
"""Connect to the IMAP server and start polling for new messages."""
try:
Expand All @@ -232,6 +253,8 @@ async def connect(self) -> bool:
if status == "OK" and data and data[0]:
for uid in data[0].split():
self._seen_uids.add(uid)
# Keep only the most recent UIDs to prevent unbounded growth
self._trim_seen_uids()
imap.logout()
logger.info("[Email] IMAP connection test passed. %d existing messages skipped.", len(self._seen_uids))
except Exception as e:
Expand Down Expand Up @@ -302,6 +325,9 @@ def _fetch_new_messages(self) -> List[Dict[str, Any]]:
if uid in self._seen_uids:
continue
self._seen_uids.add(uid)
# Trim periodically to prevent unbounded memory growth
if len(self._seen_uids) > self._seen_uids_max:
self._trim_seen_uids()

status, msg_data = imap.uid("fetch", uid, "(RFC822)")
if status != "OK":
Expand Down
Loading