fix(outlook): dispatch Contacts by item type and harden folder/field assumptions

app/connectors_service/connectors/sources/outlook/client.py

@@ -28,7 +28,7 @@
 from connectors.sources.outlook.constants import (
     API_SCOPE,
     CALENDAR_FIELDS,
-    CONTACT_FIELDS,
+    CONTACT_FOLDER_FIELDS,
     EWS_ENDPOINT,
     MAIL_FIELDS,
     MAIL_TYPES,
@@ -226,7 +226,7 @@ async def get_user_accounts(self):
         )
 
         async for user in self.get_users():
-            if "searchResRef" in user["type"]:
+            if "searchResRef" in user.get("type", ""):
                 continue
 
             mail = _extract_ldap_mail(user.get("attributes", {}))
@@ -423,20 +423,42 @@ async def get_mails(self, account):
                 yield mail, mail_type
 
     async def get_calendars(self, account):
-        for calendar in await asyncio.to_thread(
-            account.calendar.all().only, *CALENDAR_FIELDS
-        ):
+        # Skip instead of aborting if the mailbox has no Calendar folder.
+        try:
+            folder = account.calendar
+        except ErrorFolderNotFound:
+            self._logger.warning(
+                f"Could not resolve Calendar folder for {account.primary_smtp_address}, skipping."
+            )
+            return
+        for calendar in await asyncio.to_thread(folder.all().only, *CALENDAR_FIELDS):
             yield calendar
 
     async def get_child_calendars(self, account):
-        for child_calendar in account.calendar.children:
+        try:
+            child_calendars = account.calendar.children
+        except ErrorFolderNotFound:
+            self._logger.warning(
+                f"Could not resolve Calendar folder for {account.primary_smtp_address}, "
+                "skipping child calendars."
+            )
+            return
+        for child_calendar in child_calendars:
             for calendar in await asyncio.to_thread(
                 child_calendar.all().only, *CALENDAR_FIELDS
             ):
                 yield calendar, child_calendar
 
     async def get_tasks(self, account):
-        for task in await asyncio.to_thread(account.tasks.all().only, *TASK_FIELDS):
+        # Skip instead of aborting if the mailbox has no Tasks folder.
+        try:
+            folder = account.tasks
+        except ErrorFolderNotFound:
+            self._logger.warning(
+                f"Could not resolve Tasks folder for {account.primary_smtp_address}, skipping."
+            )
+            return
+        for task in await asyncio.to_thread(folder.all().only, *TASK_FIELDS):
             yield task
 
     async def get_contacts(self, account):
@@ -449,5 +471,7 @@ async def get_contacts(self, account):
                 f"Could not resolve Contacts folder for {account.primary_smtp_address}, skipping."
             )
             return
-        for contact in await asyncio.to_thread(folder.all().only, *CONTACT_FIELDS):
+        for contact in await asyncio.to_thread(
+            folder.all().only, *CONTACT_FOLDER_FIELDS
+        ):
             yield contact

app/connectors_service/connectors/sources/outlook/constants.py

@@ -71,6 +71,13 @@
     "company_name",
     "birthday",
 ]
+DISTRIBUTION_LIST_FIELDS = [
+    "last_modified_time",
+    "display_name",
+    "members",
+]
+# Contacts folder holds both item types, so query the union of their fields.
+CONTACT_FOLDER_FIELDS = list(dict.fromkeys(CONTACT_FIELDS + DISTRIBUTION_LIST_FIELDS))
 TASK_FIELDS = [
     "last_modified_time",
     "due_date",

app/connectors_service/connectors/sources/outlook/datasource.py

@@ -14,6 +14,7 @@
     iso_utc,
 )
 from exchangelib.errors import ErrorNonExistentMailbox
+from exchangelib.items import DistributionList
 
 from connectors.access_control import ACCESS_CONTROL, es_access_control_query
 from connectors.sources.outlook.client import OutlookClient, _extract_ldap_mail
@@ -34,6 +35,11 @@
 )
 from connectors.utils import html_to_text
 
+# Per-item errors: a single malformed item is skipped, not fatal. Connection-wide
+# failures (transport, SSL, rate limit, missing mailbox) are excluded on purpose
+# so they still abort the sync instead of silently emptying the index.
+ITEM_SHAPE_ERRORS = (AttributeError, KeyError, ValueError, TypeError)
+
 
 class OutlookDocFormatter:
     """Format Outlook object documents to Elasticsearch document"""
@@ -90,11 +96,13 @@ def calendar_doc_formatter(self, calendar, child_calendar, timezone):
         }
 
         if child_calendar in ["Folder (Birthdays)", "Birthdays (Birthdays)"]:
+            # calendar.start may be missing; guard against splitting a None.
+            birthday = ews_format_to_datetime(
+                source_datetime=calendar.start, timezone=timezone
+            )
             document.update(
                 {
-                    "date": ews_format_to_datetime(
-                        source_datetime=calendar.start, timezone=timezone
-                    ).split("T", 1)[0],
+                    "date": birthday.split("T", 1)[0] if birthday else None,
                 }
             )
         else:
@@ -168,6 +176,23 @@ def contact_doc_formatter(self, contact, timezone):
             ),
         }
 
+    def distribution_list_doc_formatter(self, distribution_list, timezone):
+        # A contact group has members, not per-contact fields, so it needs its own shape.
+        return {
+            "_id": distribution_list.id,
+            "type": "Distribution List",
+            "_timestamp": ews_format_to_datetime(
+                source_datetime=distribution_list.last_modified_time,
+                timezone=timezone,
+            ),
+            "name": distribution_list.display_name,
+            "email_addresses": [
+                member.mailbox.email_address
+                for member in (distribution_list.members or [])
+                if member and member.mailbox and member.mailbox.email_address
+            ],
+        }
+
     def attachment_doc_formatter(self, attachment, attachment_type, timezone):
         attachment_id = (
             attachment.attachment_id.id if attachment.attachment_id else None
@@ -488,6 +513,31 @@ async def download_func(self, content):
         """
         yield content
 
+    def _format_item(self, formatter, item, account):
+        """Run a document formatter for a single item, isolating item-shape
+        errors so one malformed Exchange object is skipped (with a warning)
+        instead of aborting the whole sync.
+
+        Args:
+            formatter (callable): Zero-argument callable that formats the item
+                (typically a functools.partial around a doc formatter).
+            item: The Exchange item being formatted, used for logging context.
+            account: The account the item belongs to, used for logging context.
+
+        Returns:
+            The formatted document, or None if the item was malformed and
+            skipped. Connection-wide errors are not caught here and propagate.
+        """
+        try:
+            return formatter()
+        except ITEM_SHAPE_ERRORS as error:
+            self._logger.warning(
+                f"Skipping malformed {type(item).__name__} item "
+                f"{getattr(item, 'id', 'unknown')} for "
+                f"{account.primary_smtp_address}: {error}"
+            )
+            return None
+
     async def _fetch_attachments(
         self, attachment_type, outlook_object, timezone, account
     ):
@@ -497,11 +547,18 @@ async def _fetch_attachments(
                     f"Skipping attachment without an ID on item {outlook_object.id}"
                 )
                 continue
-            document = self.doc_formatter.attachment_doc_formatter(
-                attachment=attachment,
-                attachment_type=attachment_type,
-                timezone=timezone,
+            document = self._format_item(
+                partial(
+                    self.doc_formatter.attachment_doc_formatter,
+                    attachment=attachment,
+                    attachment_type=attachment_type,
+                    timezone=timezone,
+                ),
+                item=attachment,
+                account=account,
             )
+            if document is None:
+                continue
             yield (
                 self._decorate_with_access_control(
                     document, [account.primary_smtp_address]
@@ -513,11 +570,18 @@ async def _fetch_attachments(
 
     async def _fetch_mails(self, account, timezone):
         async for mail, mail_type in self.client.get_mails(account=account):
-            document = self.doc_formatter.mails_doc_formatter(
-                mail=mail,
-                mail_type=mail_type,
-                timezone=timezone,
+            document = self._format_item(
+                partial(
+                    self.doc_formatter.mails_doc_formatter,
+                    mail=mail,
+                    mail_type=mail_type,
+                    timezone=timezone,
+                ),
+                item=mail,
+                account=account,
             )
+            if document is None:
+                continue
             yield (
                 self._decorate_with_access_control(
                     document, [account.primary_smtp_address]
@@ -537,10 +601,22 @@ async def _fetch_mails(self, account, timezone):
     async def _fetch_contacts(self, account, timezone):
         self._logger.debug(f"Fetching contacts for {account.primary_smtp_address}")
         async for contact in self.client.get_contacts(account=account):
-            document = self.doc_formatter.contact_doc_formatter(
-                contact=contact,
-                timezone=timezone,
-            )
+            # Contacts folder holds both contacts and groups; each has its own shape.
+            if isinstance(contact, DistributionList):
+                formatter = partial(
+                    self.doc_formatter.distribution_list_doc_formatter,
+                    distribution_list=contact,
+                    timezone=timezone,
+                )
+            else:
+                formatter = partial(
+                    self.doc_formatter.contact_doc_formatter,
+                    contact=contact,
+                    timezone=timezone,
+                )
+            document = self._format_item(formatter, item=contact, account=account)
+            if document is None:
+                continue
             yield (
                 self._decorate_with_access_control(
                     document, [account.primary_smtp_address]
@@ -551,9 +627,17 @@ async def _fetch_contacts(self, account, timezone):
     async def _fetch_tasks(self, account, timezone):
         self._logger.debug(f"Fetching tasks for {account.primary_smtp_address}")
         async for task in self.client.get_tasks(account=account):
-            document = self.doc_formatter.task_doc_formatter(
-                task=task, timezone=timezone
+            document = self._format_item(
+                partial(
+                    self.doc_formatter.task_doc_formatter,
+                    task=task,
+                    timezone=timezone,
+                ),
+                item=task,
+                account=account,
             )
+            if document is None:
+                continue
             yield (
                 self._decorate_with_access_control(
                     document, [account.primary_smtp_address]
@@ -597,11 +681,18 @@ async def _fetch_child_calendars(self, account, timezone):
                 yield doc
 
     async def _enqueue_calendars(self, calendar, child_calendar, timezone, account):
-        document = self.doc_formatter.calendar_doc_formatter(
-            calendar=calendar,
-            child_calendar=str(child_calendar),
-            timezone=timezone,
+        document = self._format_item(
+            partial(
+                self.doc_formatter.calendar_doc_formatter,
+                calendar=calendar,
+                child_calendar=str(child_calendar),
+                timezone=timezone,
+            ),
+            item=calendar,
+            account=account,
         )
+        if document is None:
+            return
         yield (
             self._decorate_with_access_control(
                 document, [account.primary_smtp_address]