Add src/watcher.py with IMAP IDLE FETCH handling fix

src/watcher.py

@@ -0,0 +1,111 @@
+async def watch_folder(
+    account: dict,
+    folder: str,
+    webhook_url: str,
+    api_key: str,
+    stop_event: asyncio.Event,
+):
+    backoff = 5
+    max_backoff = 60
+
+    while not stop_event.is_set():
+        client = None
+        try:
+            client = await asyncio.to_thread(get_imap_client, account)
+            await asyncio.to_thread(
+                client.select_folder,
+                folder,
+                readonly=not account.get("mark_as_read", False),
+            )
+
+            # Check IDLE support
+            if not client.has_capability("IDLE"):
+                logger.warning(
+                    "[%s/%s] Server does not support IDLE, falling back to polling",
+                    account["id"],
+                    folder,
+                )
+                await _poll_folder(
+                    client, account, folder, webhook_url, api_key, stop_event
+                )
+                return
+
+            # Record existing unseen UIDs so we only forward truly new ones
+            existing_unseen = set(await asyncio.to_thread(client.search, ["UNSEEN"]))
+            logger.info(
+                "[%s/%s] Watching for new emails via IDLE (%d existing unseen skipped)",
+                account["id"],
+                folder,
+                len(existing_unseen),
+            )
+            backoff = 5
+
+            while not stop_event.is_set():
+                await asyncio.to_thread(client.idle)
+                try:
+                    responses = await asyncio.to_thread(client.idle_check, 120)
+                except Exception:
+                    try:
+                        await asyncio.to_thread(client.idle_done)
+                    except Exception:
+                        pass
+                    break
+                await asyncio.to_thread(client.idle_done)
+
+                logger.debug(
+                    "[%s/%s] IDLE responses: %s", account["id"], folder, responses
+                )
+
+                uids = await asyncio.to_thread(client.search, ["UNSEEN"])
+                # Only forward emails that arrived after we started watching
+                new_uids = [u for u in uids if u not in existing_unseen]
+                if not new_uids:
+                    continue
+
+                # Fetch the full RFC822 payload for any newly‑seen messages.
+                # Some IMAP servers may emit malformed FETCH responses that cause
+                # ``client.fetch`` to raise a ``ValueError``. Previously this would
+                # crash the watcher loop. We now catch that error, log a warning,
+                # and skip the offending UIDs so the watcher can continue.
+                try:
+                    raw_messages = await asyncio.to_thread(
+                        client.fetch, new_uids, ["RFC822"]
+                    )
+                except ValueError as e:
+                    logger.warning(
+                        "[%s/%s] IMAP FETCH error for UIDs %s: %s — skipping",
+                        account["id"],
+                        folder,
+                        new_uids,
+                        e,
+                    )
+                    raw_messages = {}
+                for uid, data in raw_messages.items():
+                    raw = data.get(b"RFC822", b"")
+                    if not raw:
+                        continue
+                    email_data = parse_email_message(raw)
+                    await forward_to_poke(email_data, account, webhook_url, api_key)
+
+                if account.get("mark_as_read", False):
+                    await asyncio.to_thread(client.set_flags, new_uids, [b"\\Seen"])
+                existing_unseen.update(new_uids)
+
+        except asyncio.CancelledError:
+            break
+        except Exception as e:
+            logger.error(
+                "[%s/%s] Watcher error: %s (reconnecting in %ds)",
+                account["id"],
+                folder,
+                e,
+                backoff,
+            )
+            await asyncio.sleep(backoff)
+            backoff = min(backoff * 2, max_backoff)
+        finally:
+            if client:
+                try:
+                    await asyncio.to_thread(client.logout)
+                except Exception:
+                    pass