feat(zoom): sync recording permissions periodically (#15020)

backend/ee/onyx/configs/app_configs.py

@@ -156,6 +156,12 @@ ONEDRIVE_PERMISSION_GROUP_SYNC_FREQUENCY_S = int(
 ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY = int(
     os.environ.get("ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY") or 5 * 60
 )
+# In seconds, default is 6 hours: a sync lists every host's whole recording
+# history, about 170 calls per host, and reads the share settings of every
+# recording, one call each.
+ZOOM_PERMISSION_DOC_SYNC_FREQUENCY = int(
+    os.environ.get("ZOOM_PERMISSION_DOC_SYNC_FREQUENCY") or 6 * 60 * 60
+)
 
 
 ####

backend/ee/onyx/external_permissions/sync_params.py

@@ -24,6 +24,7 @@ from ee.onyx.configs.app_configs import (
     SLACK_PERMISSION_DOC_SYNC_FREQUENCY,
     TEAMS_PERMISSION_DOC_SYNC_FREQUENCY,
     TEAMS_PERMISSION_GROUP_SYNC_FREQUENCY,
+    ZOOM_PERMISSION_DOC_SYNC_FREQUENCY,
     ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY,
 )
 from ee.onyx.external_permissions.perm_sync_types import (
@@ -208,6 +209,12 @@ def _load_teams_group_sync() -> GroupSyncFuncType:
     return teams_group_sync
 
 
+def _load_zoom_doc_sync() -> DocSyncFuncType:
+    from ee.onyx.external_permissions.zoom.doc_sync import zoom_doc_sync
+
+    return zoom_doc_sync
+
+
 def _load_zoom_group_sync() -> GroupSyncFuncType:
     from ee.onyx.external_permissions.zoom.group_sync import zoom_group_sync
 
@@ -239,7 +246,7 @@ class SyncConfig(BaseModel):
     censoring_config: CensoringConfig | None = None
 
 
-# No-op doc sync: these sources set permissions while indexing instead.
+# No-op doc sync: the mock connector sets permissions while indexing instead.
 def mock_doc_sync(
     cc_pair: "ConnectorCredentialPair",  # noqa: ARG001
     fetch_all_docs_fn: FetchAllDocumentsFunction,  # noqa: ARG001
@@ -348,8 +355,8 @@ _SOURCE_TO_SYNC_CONFIG: dict[DocumentSource, SyncConfig] = {
     # Domain sign-in rules: groups filled from the account's user roster.
     DocumentSource.ZOOM: SyncConfig(
         doc_sync_config=DocSyncConfig(
-            doc_sync_frequency=DEFAULT_PERMISSION_DOC_SYNC_FREQUENCY,
-            doc_sync_func=mock_doc_sync,
+            doc_sync_frequency=ZOOM_PERMISSION_DOC_SYNC_FREQUENCY,
+            doc_sync_func=_lazy_doc_sync(_load_zoom_doc_sync),
             initial_index_should_sync=True,
         ),
         group_sync_config=GroupSyncConfig(

backend/ee/onyx/external_permissions/zoom/doc_sync.py

@@ -0,0 +1,38 @@
+from collections.abc import Generator
+
+from ee.onyx.external_permissions.perm_sync_types import (
+    FetchAllDocumentsFunction,
+    FetchAllDocumentsIdsFunction,
+)
+from ee.onyx.external_permissions.utils import credential_json, generic_doc_sync
+from onyx.access.models import ElementExternalAccess
+from onyx.configs.constants import DocumentSource
+from onyx.connectors.factory import build_connector_kwargs
+from onyx.connectors.zoom.connector import ZoomConnector
+from onyx.db.models import ConnectorCredentialPair
+from onyx.indexing.indexing_heartbeat import IndexingHeartbeatInterface
+
+ZOOM_DOC_SYNC_LABEL = "zoom_doc_sync"
+
+
+def zoom_doc_sync(
+    cc_pair: ConnectorCredentialPair,
+    fetch_all_existing_docs_fn: FetchAllDocumentsFunction,  # noqa: ARG001
+    fetch_all_existing_docs_ids_fn: FetchAllDocumentsIdsFunction,
+    callback: IndexingHeartbeatInterface | None,
+) -> Generator[ElementExternalAccess, None, None]:
+    connector = ZoomConnector(
+        **build_connector_kwargs(
+            cc_pair.connector.source, cc_pair.connector.connector_specific_config
+        )
+    )
+    connector.load_credentials(credential_json(cc_pair))
+
+    yield from generic_doc_sync(
+        cc_pair=cc_pair,
+        fetch_all_existing_docs_ids_fn=fetch_all_existing_docs_ids_fn,
+        callback=callback,
+        doc_source=DocumentSource.ZOOM,
+        slim_connector=connector,
+        label=ZOOM_DOC_SYNC_LABEL,
+    )

backend/ee/onyx/external_permissions/zoom/group_sync.py

@@ -13,6 +13,7 @@ from fnmatch import fnmatchcase
 from ee.onyx.db.external_perm import ExternalUserGroup
 from ee.onyx.external_permissions.utils import credential_json
 from onyx.access.utils import build_domain_group_id
+from onyx.connectors.factory import build_connector_kwargs
 from onyx.connectors.models import ConnectorMissingCredentialError
 from onyx.connectors.zoom.client import MAX_LISTING_PAGES, ZoomClient
 from onyx.connectors.zoom.connector import ZoomConnector
@@ -28,7 +29,11 @@ def zoom_group_sync(
     tenant_id: str,  # noqa: ARG001
     cc_pair: ConnectorCredentialPair,
 ) -> Generator[ExternalUserGroup, None, None]:
-    connector = ZoomConnector(**cc_pair.connector.connector_specific_config)
+    connector = ZoomConnector(
+        **build_connector_kwargs(
+            cc_pair.connector.source, cc_pair.connector.connector_specific_config
+        )
+    )
     connector.load_credentials(credential_json(cc_pair))
     if connector.client is None:
         raise ConnectorMissingCredentialError("Zoom")

backend/onyx/connectors/zoom/connector.py

@@ -9,6 +9,7 @@ its document id, with no discovery and no checkpoint.
 """
 
 import copy
+import logging
 from collections.abc import Generator
 from functools import partial
 from typing import Any
@@ -24,6 +25,7 @@ from onyx.connectors.interfaces import (
     Resolver,
     SecondsSinceUnixEpoch,
     SlimConnector,
+    SlimConnectorWithPermSync,
 )
 from onyx.connectors.models import (
     ConnectorCheckpoint,
@@ -59,7 +61,10 @@ from onyx.connectors.zoom.recordings.processing import (
     process_occurrence,
 )
 from onyx.connectors.zoom.recordings.recording_access import (
+    AccessResolver,
     RuleGrant,
+    ZoomAccessListUnavailable,
+    ZoomRecordingGone,
     load_rule_grants,
     look_up_owner_email,
     resolve_recording_access,
@@ -165,7 +170,10 @@ class ZoomConnectorCheckpoint(ConnectorCheckpoint):
 
 
 class ZoomConnector(
-    CheckpointedConnectorWithPermSync[ZoomConnectorCheckpoint], SlimConnector, Resolver
+    CheckpointedConnectorWithPermSync[ZoomConnectorCheckpoint],
+    SlimConnector,
+    SlimConnectorWithPermSync,
+    Resolver,
 ):
     def __init__(
         self,
@@ -199,6 +207,9 @@ class ZoomConnector(
         # permission-sync probe asks about the same things instead of sampling
         # again. None means it has not run.
         self._probe_sample: ProbeSample | None = None
+        # Recordings the running doc sync made private, by reason.
+        self._unnamed = 0
+        self._gone = 0
 
     def load_credentials(self, credentials: dict[str, Any]) -> dict[str, Any] | None:
         account_id = credentials.get("zoom_account_id")
@@ -285,7 +296,9 @@ class ZoomConnector(
             raise ConnectorMissingCredentialError("Zoom")
         host_id = recording.host_id
         if host_id not in self._owner_emails:
-            self._owner_emails[host_id] = look_up_owner_email(client, host_id)
+            self._owner_emails[host_id] = (
+                look_up_owner_email(client, host_id) if host_id else None
+            )
         return resolve_recording_access(
             client,
             recording,
@@ -299,6 +312,28 @@ class ZoomConnector(
             self._rule_grants = load_rule_grants(client)
         return self._rule_grants.get(rule_id)
 
+    def _synced_access(self, recording: ZoomRecordingEntry) -> ExternalAccess:
+        """Resolve one recording's access for the periodic doc sync.
+
+        Indexing can fail one document and carry on, but the doc sync cannot.
+        So when nobody can be named to read the recording, or Zoom has deleted
+        it since it was listed, this answers with an empty access list: nobody
+        may read it. Any other error is raised unchanged, which fails the
+        attempt. The shared sync writes each access list as it is yielded, so
+        the documents reached before the error keep the access this attempt
+        found and the rest keep what the last attempt wrote, until an attempt
+        gets through. The client has already retried anything transient.
+        """
+        try:
+            return self._resolve_access(recording)
+        except ZoomAccessListUnavailable as e:
+            logger.warning("%s", e)
+            self._unnamed += 1
+        except ZoomRecordingGone as e:
+            logger.info("%s", e)
+            self._gone += 1
+        return ExternalAccess.empty()
+
     def validate_checkpoint_json(self, checkpoint_json: str) -> ZoomConnectorCheckpoint:
         return ZoomConnectorCheckpoint.model_validate_json(checkpoint_json)
 
@@ -329,18 +364,41 @@ class ZoomConnector(
     ) -> GenerateSlimDocumentOutput:
         """Pruning deletes every indexed document this does not list, so the poll
         window is ignored and the callback is not needed: the caller drives its
-        own heartbeat off the batches.
+        own heartbeat off the batches."""
+        return self._slim_documents(resolve_access=None)
 
-        Not SlimConnectorWithPermSync. Zoom builds its access lists while
-        indexing, so pruning wants ids and nothing else.
-        """
+    def retrieve_all_slim_docs_perm_sync(
+        self,
+        start: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
+        end: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
+        callback: IndexingHeartbeatInterface | None = None,  # noqa: ARG002
+    ) -> GenerateSlimDocumentOutput:
+        """The doc sync makes every indexed document this does not list private,
+        so like pruning this ignores the poll window."""
+        self._unnamed = self._gone = 0
+        yield from self._slim_documents(resolve_access=self._synced_access)
+        if self._unnamed or self._gone:
+            # One line per walk, so a mass revocation is not lost among the
+            # per-recording lines. Deletions alone are routine.
+            logger.log(
+                logging.WARNING if self._unnamed else logging.INFO,
+                "Zoom permission sync made %s recording(s) private: %s that "
+                "nobody could be named for and %s that Zoom deleted during the walk",
+                self._unnamed + self._gone,
+                self._unnamed,
+                self._gone,
+            )
+
+    def _slim_documents(
+        self, resolve_access: AccessResolver | None
+    ) -> GenerateSlimDocumentOutput:
         if self.client is None:
             raise ConnectorMissingCredentialError("Zoom")
         # Checked again here because instantiate_connector skips
         # validate_connector_settings, so a connector saved with a blank form
         # would reach this, list nothing, and delete everything it indexed.
         self._raise_if_nothing_is_in_scope()
-        return zoom_slim_documents(self.client, self._sources)
+        return zoom_slim_documents(self.client, self._sources, resolve_access)
 
     def reindex(
         self,

backend/onyx/connectors/zoom/recordings/discovery.py

@@ -31,6 +31,7 @@ from onyx.connectors.zoom.recordings.models import (
     Host,
     HostScope,
     OccurrenceWork,
+    ProvenRecording,
     ZoomListingIncomplete,
     ZoomSessionType,
     fails_the_whole_run,
@@ -179,10 +180,10 @@ class DiscoveryStepResult(BaseModel):
 
 class InventoryScope(BaseModel):
     hosts: list[HostScope] = Field(default_factory=list)
-    # Occurrences Zoom answered for directly, kept whatever a host listing goes
+    # Recordings Zoom answered for directly, kept whatever a host listing goes
     # on to say later. A listing reaches neither a recording made on-premise nor
     # one whose owner nobody can resolve.
-    proven: list[OccurrenceWork] = Field(default_factory=list)
+    proven: list[ProvenRecording] = Field(default_factory=list)
     # Hosts and sessions Zoom has no record of. Tracked so the walk can tell
     # one that was deleted from a credential pointed at the wrong Zoom account,
     # which makes every one of them look deleted.
@@ -339,16 +340,10 @@ class IdAllowlistSource(DiscoverySource):
             handler = get_session_type_handler(session_type)
             found = handler.find_host(client, session_id)
 
-            proven: list[OccurrenceWork] = []
+            proven: list[ProvenRecording] = []
             if found.anchor is not None:
                 proven.append(
-                    OccurrenceWork(
-                        session_type=session_type,
-                        session_id=session_id,
-                        occurrence_uuid=found.anchor.uuid,
-                        start_time=found.anchor.start_time,
-                        topic=found.anchor.topic,
-                    )
+                    ProvenRecording(session_type=session_type, recording=found.anchor)
                 )
 
             if found.host_id is not None and found.host_id not in listable:

backend/onyx/connectors/zoom/recordings/inventory.py

@@ -1,14 +1,11 @@
-"""Every Zoom document that still exists, for pruning.
-
-A document still exists if its recording still appears in its host's recordings
-listing, which leaves deleted and trashed recordings out and has no age limit.
-
-This is the mirror image of discovery.py. Discovery skips what it cannot read
-and reports a failure, but anything skipped here is deleted instead, so this
-raises rather than skipping, and it walks from Zoom's launch rather than the
-poll window. The one thing it does skip is a host Zoom says it has no record
-of, whose documents are then meant to go. The caller renews its lock only when
-it receives a batch, so batches go out even when they are empty.
+"""Every Zoom document that still exists, for pruning and the permission doc
+sync: one whose recording its host's listing still shows, however old, which
+leaves deleted and trashed recordings out. Anything left out here is deleted by
+pruning and made private by the doc sync, so unlike discovery this raises rather
+than skips, and it walks from Zoom's launch rather than the poll window. The one
+skip is a host Zoom says it has no record of, whose documents are meant to go.
+The caller renews its lock only when it receives a batch, so batches go out
+even when they are empty.
 """
 
 from collections.abc import Iterator
@@ -26,8 +23,9 @@ from onyx.connectors.zoom.recordings.discovery import (
     list_every_recording,
     listing_windows,
 )
-from onyx.connectors.zoom.recordings.models import Host, HostScope, OccurrenceWork
+from onyx.connectors.zoom.recordings.models import Host, HostScope, ZoomSessionType
 from onyx.connectors.zoom.recordings.processing import zoom_document_id
+from onyx.connectors.zoom.recordings.recording_access import AccessResolver
 from onyx.connectors.zoom.recordings.session_types import (
     is_portal_upload,
     session_type_for_recording,
@@ -37,6 +35,9 @@ from onyx.utils.logger import setup_logger
 logger = setup_logger()
 
 _MAX_DOCUMENTS_PER_BATCH = 500
+# Bounds the ids remembered to skip a recording listed twice, at about 30 MB.
+# Past it a repeat listing costs one more access call, and nothing is lost.
+_MAX_REMEMBERED_IDS = 250_000
 # listing_windows runs a day past the end it is given, so 28 is the widest
 # trailing pass that still fits in one Zoom call per host.
 _TRAILING_WINDOW_DAYS = 28
@@ -46,27 +47,38 @@ _SCOPES_PER_HEARTBEAT = 25
 
 
 def zoom_slim_documents(
-    client: ZoomClient, sources: list[DiscoverySource]
+    client: ZoomClient,
+    sources: list[DiscoverySource],
+    resolve_access: AccessResolver | None,
 ) -> GenerateSlimDocumentOutput:
     """Rebuilds each batch on the way out. Onyx types a slim batch as a list that
     may hold a HierarchyNode, and a list is invariant, so this copy is what lets
     everything below say what it really produces."""
-    for batch in _slim_batches(client, sources):
+    for batch in _slim_batches(client, sources, resolve_access):
         yield [*batch]
 
 
 def _slim_batches(
-    client: ZoomClient, sources: list[DiscoverySource]
+    client: ZoomClient,
+    sources: list[DiscoverySource],
+    resolve_access: AccessResolver | None,
 ) -> Iterator[list[SlimDocument]]:
     scopes: list[HostScope] = []
     proven: list[SlimDocument] = []
+    emitted: set[str] = set()
     unrecognised = 0
     anchors = 0
     resolved = 0
     for source in sources:
         for scope in source.inventory_scopes(client):
             scopes.extend(scope.hosts)
-            proven.extend(_slim_document(work) for work in scope.proven)
+            proven.extend(
+                document
+                for p in scope.proven
+                for document in _documents(
+                    {p.session_type}, p.recording, resolve_access, emitted
+                )
+            )
             unrecognised += len(scope.unrecognised)
             anchors += len(scope.proven)
             resolved += 1
@@ -77,6 +89,12 @@ def _slim_batches(
         yield proven
 
     hosts = _merged(scopes)
+    # The same walk serves pruning and the permission sync.
+    job, consequence = (
+        ("pruning", "delete every document it has indexed")
+        if resolve_access is None
+        else ("permission sync", "make every document it has indexed private")
+    )
     if unrecognised and not hosts and not anchors:
         # Zoom answers the same not-found for a user or session that was deleted
         # and for one in another account, so a credential pointed at the wrong
@@ -84,15 +102,16 @@ def _slim_batches(
         # One recording Zoom did answer for proves the account is right.
         raise ConnectorValidationError(
             "Zoom recognised none of the hosts or sessions this connector names, "
-            "so pruning stopped rather than delete every document it has "
-            "indexed. Either they were all deleted in Zoom, or the credentials "
-            "now point at a different account. To prune them anyway, replace "
-            "the entries Zoom no longer has or delete the connector"
+            f"so {job} stopped rather than {consequence}. Either they were all "
+            "deleted in Zoom, or the credentials now point at a different "
+            "account. To go ahead anyway, replace the entries Zoom no longer has "
+            "or delete the connector"
         )
     today = datetime.now(timezone.utc).date()
     windows = listing_windows(EARLIEST_RECORDING_DATE, today)
     logger.info(
-        "Zoom pruning is listing %s host(s) over %s windows, about %s calls",
+        "Zoom %s is listing %s host(s) over %s windows, about %s calls",
+        job,
         len(hosts),
         len(windows),
         len(hosts) * (len(windows) + 1),
@@ -100,7 +119,9 @@ def _slim_batches(
 
     unrecognised_types: set[str] = set()
     for scope in hosts:
-        yield from _host_documents(client, scope, windows, unrecognised_types)
+        yield from _host_documents(
+            client, scope, windows, unrecognised_types, resolve_access, emitted
+        )
 
     # A recording that finishes while this walk runs lands in the newest window,
     # so every host is asked for that window again once the walk is over. This
@@ -112,7 +133,9 @@ def _slim_batches(
     now = datetime.now(timezone.utc).date()
     trailing = listing_windows(now - timedelta(days=_TRAILING_WINDOW_DAYS), now)
     for scope in hosts:
-        yield from _host_documents(client, scope, trailing, unrecognised_types)
+        yield from _host_documents(
+            client, scope, trailing, unrecognised_types, resolve_access, emitted
+        )
 
 
 def _merged(scopes: list[HostScope]) -> list[HostScope]:
@@ -147,6 +170,8 @@ def _host_documents(
     scope: HostScope,
     windows: list[tuple[date, date]],
     unrecognised_types: set[str],
+    resolve_access: AccessResolver | None,
+    emitted: set[str],
 ) -> Iterator[list[SlimDocument]]:
     """Do not reach for `_UserRecordingsSource._recordings` instead. It trims to
     the poll window, so it would drop every meeting that ran while this walk was
@@ -155,7 +180,11 @@ def _host_documents(
 
     for from_date, to_date in windows:
         for recording in list_every_recording(client, scope.host, from_date, to_date):
-            batch.extend(_slim_documents_for(recording, scope, unrecognised_types))
+            batch.extend(
+                _slim_documents_for(
+                    recording, scope, unrecognised_types, resolve_access, emitted
+                )
+            )
             if len(batch) >= _MAX_DOCUMENTS_PER_BATCH:
                 yield batch
                 batch = []
@@ -164,7 +193,11 @@ def _host_documents(
 
 
 def _slim_documents_for(
-    recording: ZoomRecordingEntry, scope: HostScope, unrecognised_types: set[str]
+    recording: ZoomRecordingEntry,
+    scope: HostScope,
+    unrecognised_types: set[str],
+    resolve_access: AccessResolver | None,
+    emitted: set[str],
 ) -> list[SlimDocument]:
     if recording.type is not None and is_portal_upload(recording.type):
         return []
@@ -186,25 +219,35 @@ def _slim_documents_for(
             recording.uuid,
         )
 
+    document_types = scope.emitted_types(recording.session_id, derived)
+    return _documents(document_types, recording, resolve_access, emitted)
+
+
+def _documents(
+    document_types: set[ZoomSessionType],
+    recording: ZoomRecordingEntry,
+    resolve_access: AccessResolver | None,
+    emitted: set[str],
+) -> list[SlimDocument]:
+    """One document per type indexing may have written this recording under,
+    normally one. A recording the trailing pass or a walked host lists again
+    is skipped, so it costs no second access call."""
+    new_ids = [
+        document_id
+        for document_type in document_types
+        if (document_id := zoom_document_id(document_type, recording.uuid))
+        not in emitted
+    ]
+    if not new_ids:
+        return []
+    if len(emitted) < _MAX_REMEMBERED_IDS:
+        emitted.update(new_ids)
+    access = resolve_access(recording) if resolve_access is not None else None
+    created_at = _created_at(recording.start_time)
     return [
-        _slim_document(
-            OccurrenceWork(
-                session_type=session_type,
-                session_id=recording.session_id,
-                occurrence_uuid=recording.uuid,
-                start_time=recording.start_time,
-                topic=recording.topic,
-            )
-        )
-        for session_type in scope.emitted_types(recording.session_id, derived)
+        SlimDocument(id=document_id, doc_created_at=created_at, external_access=access)
+        for document_id in new_ids
     ]
-
-
-def _slim_document(work: OccurrenceWork) -> SlimDocument:
-    return SlimDocument(
-        id=zoom_document_id(work.session_type, work.occurrence_uuid),
-        doc_created_at=_created_at(work.start_time),
-    )
 
 
 def _created_at(start_time: str | None) -> datetime | None:

backend/onyx/connectors/zoom/recordings/models.py

@@ -44,15 +44,21 @@ class OccurrenceWork(BaseModel):
 
 
 class HostScope(BaseModel):
-    """The host and Group mechanisms scope by `session_types`, keeping every
-    recording of those types. The ID allowlist scopes by `sessions`, keeping
-    those numbers whatever else the host recorded."""
+    """Which of one host's recordings to keep, and as which document type."""
 
     host: Host
+    # Set when the admin listed this host by email or through a Zoom Group.
+    # Holds the ticked checkboxes, meetings and/or webinars: keep every
+    # recording of those kinds.
     session_types: frozenset[ZoomSessionType] = frozenset()
+    # Set when the admin typed a meeting or webinar number this host owns.
+    # Each entry is (the field it was typed into, the number), for example
+    # (MEETING, "111"): keep that number's recordings as meetings.
     sessions: frozenset[tuple[ZoomSessionType, str]] = frozenset()
 
     def merged_with(self, other: "HostScope") -> "HostScope":
+        """Join the scopes of a host that was both listed by email or Group and
+        named by a number, so the host is walked once."""
         return HostScope(
             host=Host(
                 user_id=self.host.user_id,
@@ -62,25 +68,30 @@ class HostScope(BaseModel):
             sessions=self.sessions | other.sessions,
         )
 
+    # TODO(subash): reject a number typed into the wrong field at validation and
+    # collapse this to one type. Zoom lists a webinar's occurrences on
+    # /past_meetings/{id}/instances, so a webinar number in the meeting field is
+    # indexed under the wrong type today; /meetings/{id} answers such a number
+    # with 400 code 3000, which is the check. Checked live on 2026-09-23.
     def emitted_types(
         self, session_id: str, derived: ZoomSessionType | None
     ) -> set[ZoomSessionType]:
-        """Every session type any scope claims for this recording.
+        """Which document types to make for one recording.
 
-        Generous on purpose, because an id the index never held costs nothing
-        while a missing one deletes a document. That covers a number configured
-        as both a meeting and a webinar, a number typed into the wrong field,
-        and a type code Zoom adds after this was written."""
-        claimed = {
+        The recording is kept as a meeting if the admin typed its number into
+        the meeting field, and likewise for webinars. It is also kept as its
+        own kind, `derived`, when that checkbox is ticked. When Zoom's type
+        code is one this connector cannot read, `derived` is None and the
+        recording is kept as every ticked kind: an extra id costs nothing,
+        while a missing one deletes a document."""
+        typed_as = {
             session_type
-            for session_type, claimed_id in self.sessions
-            if claimed_id == session_id
+            for session_type, number in self.sessions
+            if number == session_id
         }
         if derived is None:
-            return claimed | set(self.session_types)
-        if derived in self.session_types:
-            claimed.add(derived)
-        return claimed
+            return typed_as | set(self.session_types)
+        return typed_as | ({derived} & self.session_types)
 
 
 class SessionHost(BaseModel):
@@ -94,6 +105,14 @@ class SessionHost(BaseModel):
     anchor: ZoomRecordingEntry | None = None
 
 
+class ProvenRecording(BaseModel):
+    """A recording Zoom answered for by its session number, under the type the
+    admin typed the number into."""
+
+    session_type: ZoomSessionType
+    recording: ZoomRecordingEntry
+
+
 class RecordingsState(BaseModel):
     source_index: int = 0
     # Only the active source may read what is inside this. The connector

backend/onyx/connectors/zoom/recordings/processing.py

@@ -3,9 +3,6 @@ Anything that differs between meetings and webinars belongs on the
 SessionTypeHandler, not in a branch here.
 """
 
-from collections.abc import Callable
-
-from onyx.access.models import ExternalAccess
 from onyx.configs.constants import DocumentSource
 from onyx.connectors.cross_connector_utils.miscellaneous_utils import time_str_to_utc
 from onyx.connectors.models import (
@@ -15,14 +12,16 @@ from onyx.connectors.models import (
     TextSection,
 )
 from onyx.connectors.zoom.client import ZoomClient
-from onyx.connectors.zoom.models import ZoomRecordingEntry
 from onyx.connectors.zoom.recordings.models import (
     OccurrenceWork,
     ZoomSessionType,
     fails_the_whole_run,
     has_no_transcript,
 )
-from onyx.connectors.zoom.recordings.recording_access import ZoomAccessListUnavailable
+from onyx.connectors.zoom.recordings.recording_access import (
+    AccessResolver,
+    ZoomAccessListUnavailable,
+)
 from onyx.connectors.zoom.recordings.session_types import get_session_type_handler
 from onyx.file_processing.webvtt import parse_vtt_transcript
 from onyx.utils.logger import setup_logger
@@ -63,7 +62,7 @@ def process_occurrence(
     client: ZoomClient,
     work: OccurrenceWork,
     *,
-    resolve_access: Callable[[ZoomRecordingEntry], ExternalAccess] | None,
+    resolve_access: AccessResolver | None,
 ) -> Document | ConnectorFailure | None:
     """One occurrence is at most one transcript, so this answers with the
     document, the failure that replaces it, or nothing when the occurrence has

backend/onyx/connectors/zoom/recordings/recording_access.py

@@ -13,6 +13,7 @@ catalogue is only asked for when a recording names a rule.
 from collections.abc import Callable, Sequence
 from typing import NamedTuple
 
+import requests
 from pydantic import ValidationError
 
 from onyx.access.models import ExternalAccess
@@ -27,11 +28,16 @@ from onyx.connectors.zoom.models import (
     ZoomRecordingSettings,
     ZoomShareRecording,
 )
-from onyx.connectors.zoom.recordings.models import user_does_not_exist
+from onyx.connectors.zoom.recordings.models import (
+    definitely_absent,
+    user_does_not_exist,
+)
 from onyx.utils.logger import setup_logger
 
 logger = setup_logger()
 
+AccessResolver = Callable[[ZoomRecordingEntry], ExternalAccess]
+
 # Zoom's `authentication_option` for "Only people with access". It never appears
 # in the rule catalogue, and it arrives with `share_recording` still "publicly",
 # so it is checked before that field is read.
@@ -99,6 +105,12 @@ class ZoomAccessListUnavailable(Exception):
     the whole attempt."""
 
 
+class ZoomRecordingGone(Exception):
+    """The recording's own settings answered not found, so it was deleted after
+    it was listed. A not-found from any other call says nothing about the
+    recording, so it is not turned into this."""
+
+
 def approved_registrant_emails(
     registrants: Sequence[ZoomRecordingRegistrant],
 ) -> list[str]:
@@ -147,7 +159,7 @@ def resolve_recording_access(
     """Raises ZoomAccessListUnavailable rather than answering with an empty
     list, which would read as nobody having access."""
     try:
-        settings = client.get_recording_settings(recording.uuid)
+        settings = _recording_settings(client, recording.uuid)
         grant = _link_access(
             settings, recording.uuid, treat_link_access_as_public, rule_grant
         )
@@ -205,6 +217,17 @@ def resolve_recording_access(
     return access
 
 
+def _recording_settings(client: ZoomClient, uuid: str) -> ZoomRecordingSettings:
+    try:
+        return client.get_recording_settings(uuid)
+    except requests.HTTPError as e:
+        if definitely_absent(e):
+            raise ZoomRecordingGone(
+                f"Zoom recording {uuid} is gone since it was listed"
+            ) from e
+        raise
+
+
 def _link_access(
     settings: ZoomRecordingSettings,
     uuid: str,

backend/tests/unit/ee/onyx/external_permissions/zoom/test_zoom_doc_sync.py

@@ -0,0 +1,85 @@
+"""The Zoom doc sync hands the connector's permission-aware walk to the shared
+sync, which turns it into access rows and hides what the walk no longer lists."""
+
+from unittest.mock import MagicMock, patch
+
+from ee.onyx.configs.app_configs import ZOOM_PERMISSION_DOC_SYNC_FREQUENCY
+from ee.onyx.external_permissions.sync_params import get_source_perm_sync_config
+from ee.onyx.external_permissions.zoom.doc_sync import zoom_doc_sync
+from onyx.access.models import DocExternalAccess, ExternalAccess
+from onyx.configs.constants import DocumentSource
+from onyx.connectors.models import SlimDocument
+
+MODULE = "ee.onyx.external_permissions.zoom.doc_sync"
+
+OWNER = ExternalAccess(
+    external_user_emails={"owner@example.com"},
+    external_user_group_ids=set(),
+    is_public=False,
+)
+
+
+def _cc_pair() -> MagicMock:
+    cc_pair = MagicMock()
+    cc_pair.id = 7
+    cc_pair.connector.connector_specific_config = {"host_emails": ["owner@example.com"]}
+    cc_pair.connector.indexing_start = None
+    cc_pair.credential.credential_json.get_value.return_value = {
+        "zoom_account_id": "acct"
+    }
+    return cc_pair
+
+
+def test_the_walk_becomes_access_rows_and_unlisted_documents_go_private() -> None:
+    connector = MagicMock()
+    connector.retrieve_all_slim_docs_perm_sync.return_value = iter(
+        [
+            [
+                SlimDocument(id="ZOOM_MEETING_uuid-1", external_access=OWNER),
+                SlimDocument(
+                    id="ZOOM_MEETING_uuid-2", external_access=ExternalAccess.empty()
+                ),
+            ]
+        ]
+    )
+
+    with patch(f"{MODULE}.ZoomConnector", return_value=connector) as factory:
+        rows = list(
+            zoom_doc_sync(
+                _cc_pair(),
+                MagicMock(),
+                lambda: [
+                    "ZOOM_MEETING_uuid-1",
+                    "ZOOM_MEETING_uuid-2",
+                    "ZOOM_MEETING_gone",
+                ],
+                None,
+            )
+        )
+
+    factory.assert_called_once_with(host_emails=["owner@example.com"])
+    connector.load_credentials.assert_called_once_with({"zoom_account_id": "acct"})
+    assert rows == [
+        DocExternalAccess(doc_id="ZOOM_MEETING_uuid-1", external_access=OWNER),
+        DocExternalAccess(
+            doc_id="ZOOM_MEETING_uuid-2", external_access=ExternalAccess.empty()
+        ),
+        DocExternalAccess(
+            doc_id="ZOOM_MEETING_gone", external_access=ExternalAccess.empty()
+        ),
+    ]
+
+
+def test_zoom_is_registered_with_its_own_doc_sync() -> None:
+    config = get_source_perm_sync_config(DocumentSource.ZOOM)
+    assert config is not None and config.doc_sync_config is not None
+    cc_pair, fetch_docs, fetch_ids = MagicMock(), MagicMock(), MagicMock()
+
+    # The registry holds a lazy wrapper, so calling it is what shows where it leads.
+    with patch(f"{MODULE}.zoom_doc_sync") as doc_sync:
+        config.doc_sync_config.doc_sync_func(cc_pair, fetch_docs, fetch_ids, None)
+
+    doc_sync.assert_called_once_with(cc_pair, fetch_docs, fetch_ids, None)
+    assert (
+        config.doc_sync_config.doc_sync_frequency == ZOOM_PERMISSION_DOC_SYNC_FREQUENCY
+    )

backend/tests/unit/onyx/connectors/zoom/test_zoom_slim_retrieval.py

@@ -6,6 +6,7 @@ does it raise rather than answer short.
 """
 
 import itertools
+from collections.abc import Callable
 from datetime import date, datetime, timezone
 from typing import Any
 from unittest.mock import MagicMock, patch
@@ -13,11 +14,13 @@ from unittest.mock import MagicMock, patch
 import pytest
 import requests
 
+from onyx.access.models import ExternalAccess
 from onyx.background.celery.celery_utils import extract_ids_from_runnable_connector
 from onyx.connectors.exceptions import (
     ConnectorValidationError,
     InsufficientPermissionsError,
 )
+from onyx.connectors.interfaces import GenerateSlimDocumentOutput
 from onyx.connectors.models import ConnectorMissingCredentialError, SlimDocument
 from onyx.connectors.zoom.connector import ZoomConnector
 from onyx.connectors.zoom.models import (
@@ -28,6 +31,7 @@ from onyx.connectors.zoom.models import (
     ZoomUser,
     ZoomUserPage,
 )
+from onyx.connectors.zoom.recordings import inventory
 from onyx.connectors.zoom.recordings.discovery import (
     EARLIEST_RECORDING_DATE,
     listing_windows,
@@ -36,11 +40,16 @@ from onyx.connectors.zoom.recordings.inventory import (
     _MAX_DOCUMENTS_PER_BATCH,
     _SCOPES_PER_HEARTBEAT,
 )
-from tests.unit.onyx.connectors.zoom.helpers import http_error, mock_zoom_client
+from tests.unit.onyx.connectors.zoom.helpers import (
+    http_error,
+    mock_zoom_client,
+    with_recording_access,
+)
 from tests.unit.onyx.connectors.zoom.zoom_api_shapes import (
     meeting_details,
     past_meeting_details,
     recording_entry,
+    recording_settings,
     user,
     webinar_details,
 )
@@ -54,6 +63,14 @@ _ZOOM_CREDS = {
 # The id every recording builder defaults to.
 _NUMBER = "6840331990"
 
+# What the access stub's default share settings resolve to: the owner, and
+# public because the recording is shared with the account and the box is on.
+_ACCOUNT_SHARE = ExternalAccess(
+    external_user_emails={"owner@example.com"},
+    external_user_group_ids=set(),
+    is_public=True,
+)
+
 
 def _windows_per_host() -> int:
     """What one host costs the walk. Zoom's launch is the listing floor and
@@ -101,19 +118,27 @@ def _listing(client: MagicMock, *recordings: ZoomRecordingEntry) -> None:
     client.list_user_recordings.return_value = _recording_page(*recordings)
 
 
-def _documents(connector: ZoomConnector) -> list[SlimDocument]:
+def _items(batches: GenerateSlimDocumentOutput) -> list[SlimDocument]:
     return [
-        item
-        for batch in connector.retrieve_all_slim_docs()
-        for item in batch
-        if isinstance(item, SlimDocument)
+        item for batch in batches for item in batch if isinstance(item, SlimDocument)
     ]
 
 
+def _documents(connector: ZoomConnector) -> list[SlimDocument]:
+    return _items(connector.retrieve_all_slim_docs())
+
+
 def _ids(connector: ZoomConnector) -> set[str]:
     return {document.id for document in _documents(connector)}
 
 
+def _synced(connector: ZoomConnector) -> dict[str, ExternalAccess | None]:
+    return {
+        item.id: item.external_access
+        for item in _items(connector.retrieve_all_slim_docs_perm_sync())
+    }
+
+
 def _windows(client: MagicMock) -> list[dict[str, Any]]:
     """The listing calls that were windows of the walk. The one-day probe that
     checks a host exists before the walk is not one."""
@@ -264,14 +289,24 @@ class TestSlimFailuresNeverDeleteAnything:
         assert asked.count("gone@example.com") == 1
         assert "gone@example.com" not in _walked(client)
 
-    def test_a_credential_that_recognises_nobody_stops_the_prune(self) -> None:
+    @pytest.mark.parametrize(
+        ("walk", "consequence"),
+        [
+            (_documents, "pruning stopped rather than delete"),
+            (_synced, "permission sync stopped rather than make"),
+        ],
+        ids=["pruning", "doc-sync"],
+    )
+    def test_a_credential_that_recognises_nobody_stops_the_walk(
+        self, walk: Callable[[ZoomConnector], object], consequence: str
+    ) -> None:
         # Zoom answers the same 1001 for a deleted user and for one in another
         # account, so a credential pointed elsewhere would delete everything.
         connector, client = _connector(host_emails=["jill@example.com"])
         client.list_user_recordings.side_effect = _listing_by_host({})
 
-        with pytest.raises(ConnectorValidationError, match="recognised none"):
-            list(connector.retrieve_all_slim_docs())
+        with pytest.raises(ConnectorValidationError, match=consequence):
+            walk(connector)
 
     @pytest.mark.parametrize(
         "error",
@@ -571,3 +606,168 @@ class TestBatchingKeepsThePruneAlive:
 
         assert len(batches) > 1
         assert all(len(batch) <= _MAX_DOCUMENTS_PER_BATCH for batch in batches[:-1])
+
+
+class TestThePermSyncWalk:
+    """The doc sync makes every indexed document the walk leaves out private,
+    so it lists exactly what pruning lists, with each recording's access."""
+
+    def test_every_document_carries_the_recordings_access(self) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        _listing(client, _recording("uuid-1"), _recording("uuid-2"))
+
+        synced = _synced(connector)
+
+        assert synced == {
+            "ZOOM_MEETING_uuid-1": _ACCOUNT_SHARE,
+            "ZOOM_MEETING_uuid-2": _ACCOUNT_SHARE,
+        }
+        assert set(synced) == _ids(connector)
+
+    def test_pruning_asks_for_no_access(self) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        _listing(client, _recording("uuid-1"))
+
+        documents = _documents(connector)
+
+        assert [d.external_access for d in documents] == [None]
+        client.get_recording_settings.assert_not_called()
+
+    def test_a_recording_listed_again_costs_no_second_access_call(self) -> None:
+        # The stub answers every window and the trailing pass with the same
+        # recording, so the walk lists it many times over.
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        _listing(client, _recording("uuid-1"))
+
+        _synced(connector)
+
+        client.get_recording_settings.assert_called_once()
+        client.get_user.assert_called_once()
+
+    def test_past_the_remembered_ids_cap_nothing_is_lost(self) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        _listing(client, _recording("uuid-1"), _recording("uuid-2"))
+
+        with patch.object(inventory, "_MAX_REMEMBERED_IDS", 1):
+            synced = _synced(connector)
+
+        assert set(synced) == {"ZOOM_MEETING_uuid-1", "ZOOM_MEETING_uuid-2"}
+        # The second recording was not remembered, so each listing resolved it.
+        assert client.get_recording_settings.call_count > 2
+
+    def test_a_recording_nobody_can_be_named_for_is_private(self) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(
+            client, settings=recording_settings(share_recording="none")
+        )
+        client.get_user.side_effect = http_error(404, 1001)
+        _listing(client, _recording("uuid-1"))
+
+        assert _synced(connector) == {"ZOOM_MEETING_uuid-1": ExternalAccess.empty()}
+
+    def test_the_walk_ends_with_a_count_of_what_went_private(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(
+            client, settings=recording_settings(share_recording="none")
+        )
+        client.get_user.side_effect = http_error(404, 1001)
+        _listing(client, _recording("uuid-1"), _recording("uuid-2"))
+
+        with caplog.at_level("WARNING"):
+            _synced(connector)
+
+        said = [r for r in caplog.records if "made 2 recording(s)" in r.getMessage()]
+        assert len(said) == 1
+        assert "2 that nobody could be named for" in said[0].getMessage()
+
+    def test_a_recording_deleted_mid_walk_is_private(self) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        client.get_recording_settings.side_effect = http_error(404)
+        _listing(client, _recording("uuid-1"))
+
+        assert _synced(connector) == {"ZOOM_MEETING_uuid-1": ExternalAccess.empty()}
+
+    def test_any_other_settings_error_fails_the_sync(self) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        client.get_recording_settings.side_effect = http_error(400, 1234)
+        _listing(client, _recording("uuid-1"))
+
+        with pytest.raises(requests.HTTPError):
+            _synced(connector)
+
+    @pytest.mark.parametrize(
+        "refuse",
+        [
+            lambda client: client.get_recording_authentication_rules,
+            lambda client: client.get_user,
+        ],
+        ids=["rule-catalogue", "owner-lookup"],
+    )
+    def test_a_404_from_any_other_call_fails_the_sync(
+        self, refuse: Callable[[MagicMock], MagicMock]
+    ) -> None:
+        # Only the recording's own settings answering 404 means it was deleted.
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        refuse(client).side_effect = http_error(404)
+        _listing(client, _recording("uuid-1"), _recording("uuid-2"))
+
+        with pytest.raises(requests.HTTPError):
+            _synced(connector)
+
+    def test_a_proven_recording_is_resolved_from_its_anchor(self) -> None:
+        connector, client = _connector(meeting_ids=[_NUMBER])
+        with_recording_access(client)
+        client.get_recording.return_value = _recording("uuid-proven")
+        client.list_user_recordings.side_effect = itertools.repeat(
+            http_error(404, 1001)
+        )
+
+        assert _synced(connector) == {"ZOOM_MEETING_uuid-proven": _ACCOUNT_SHARE}
+
+    def test_a_proven_recording_its_host_also_lists_is_resolved_once(self) -> None:
+        connector, client = _connector(meeting_ids=[_NUMBER])
+        with_recording_access(client)
+        recording = _recording("uuid-proven")
+        client.get_recording.return_value = recording
+        _listing(client, recording)
+
+        assert _synced(connector) == {"ZOOM_MEETING_uuid-proven": _ACCOUNT_SHARE}
+        client.get_recording_settings.assert_called_once()
+
+    def test_a_recording_with_no_host_is_resolved_without_an_owner(self) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        _listing(client, _recording("uuid-1", host_id=""))
+
+        synced = _synced(connector)
+
+        assert synced == {
+            "ZOOM_MEETING_uuid-1": ExternalAccess(
+                external_user_emails=set(),
+                external_user_group_ids=set(),
+                is_public=True,
+            )
+        }
+        client.get_user.assert_not_called()
+
+    def test_a_recording_of_unreadable_type_carries_one_access_on_every_id(
+        self,
+    ) -> None:
+        connector, client = _connector(host_emails=["jill@example.com"])
+        with_recording_access(client)
+        _listing(client, _recording("uuid-1", type="4242"))
+
+        assert _synced(connector) == {
+            "ZOOM_MEETING_uuid-1": _ACCOUNT_SHARE,
+            "ZOOM_WEBINAR_uuid-1": _ACCOUNT_SHARE,
+        }
+        client.get_recording_settings.assert_called_once()

cli/internal/deploy/deployfiles/embedded/docker_compose/env.template

@@ -412,6 +412,7 @@ LOG_ONYX_MODEL_INTERACTIONS=False
 # GONG_CONNECTOR_START_TIME=
 # ZOOM_TRANSCRIPT_LAG_BUFFER_HOURS=
 # ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY=
+# ZOOM_PERMISSION_DOC_SYNC_FREQUENCY=
 # NOTION_CONNECTOR_DISABLE_RECURSIVE_PAGE_LOOKUP=
 # GITHUB_CONNECTOR_BASE_URL=
 # MAX_DOCUMENT_CHARS=

deployment/docker_compose/env.template

@@ -412,6 +412,7 @@ LOG_ONYX_MODEL_INTERACTIONS=False
 # GONG_CONNECTOR_START_TIME=
 # ZOOM_TRANSCRIPT_LAG_BUFFER_HOURS=
 # ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY=
+# ZOOM_PERMISSION_DOC_SYNC_FREQUENCY=
 # NOTION_CONNECTOR_DISABLE_RECURSIVE_PAGE_LOOKUP=
 # GITHUB_CONNECTOR_BASE_URL=
 # MAX_DOCUMENT_CHARS=

deployment/helm/charts/onyx/Chart.yaml

@@ -5,7 +5,7 @@ home: https://www.onyx.app/
 sources:
   - "https://github.com/onyx-dot-app/onyx"
 type: application
-version: 0.9.7
+version: 0.9.8
 appVersion: latest
 annotations:
   category: Productivity

deployment/helm/charts/onyx/values.yaml

@@ -1781,6 +1781,7 @@ configMap:
   GONG_CONNECTOR_START_TIME: ""
   ZOOM_TRANSCRIPT_LAG_BUFFER_HOURS: ""
   ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY: ""
+  ZOOM_PERMISSION_DOC_SYNC_FREQUENCY: ""
   NOTION_CONNECTOR_DISABLE_RECURSIVE_PAGE_LOOKUP: ""
   # Push every successfully indexed public-connector document to an external
   # HTTP endpoint (failures never fail indexing). Intended for non-EE