Auto-refresh zoom tokens via celery task + implement task locks

backend/.env.example

@@ -104,6 +104,7 @@ ZOOM_API_ENABLED=False
 ZOOM_AUTH_CLIENT_ID=
 ZOOM_AUTH_SECRET=
 ZOOM_AUTH_CALLBACK=http://localhost:5000/zoom/callback
+ZOOM_TOKEN_TTL_IN_SECONDS=604800 # 7 days
 
 # -- SIGNED URL SECRET --
 # Shared secret for url signing (e.g. create it by running `openssl rand -hex 32`)
@@ -154,4 +155,4 @@ OIDC_TOKEN_INTROSPECTION_URL=
 OIDC_FALLBACK_MATCH_BY_EMAIL=
 
 # Required for Appointment's CalDAV auto-setup, needs to match the one in thunderbird-accounts
-APPOINTMENT_CALDAV_SECRET=
+APPOINTMENT_CALDAV_SECRET=

backend/.env.test

@@ -63,6 +63,7 @@ ZOOM_API_ENABLED=False
 ZOOM_AUTH_CLIENT_ID=
 ZOOM_AUTH_SECRET=
 ZOOM_AUTH_CALLBACK=http://localhost:8090/zoom/callback
+ZOOM_TOKEN_TTL_IN_SECONDS=604800 # 7 days
 
 # -- SIGNED URL SECRET --
 # Shared secret for url signing (e.g. create it by running `openssl rand -hex 32`)
@@ -117,4 +118,4 @@ OIDC_EXP_GRACE_PERIOD=300
 OIDC_FALLBACK_MATCH_BY_EMAIL=
 
 # Required for Appointment's CalDAV auto-setup, needs to match the one in thunderbird-accounts
-APPOINTMENT_CALDAV_SECRET=
+APPOINTMENT_CALDAV_SECRET=

backend/src/appointment/celery_app.py

@@ -44,6 +44,9 @@ def create_celery_app() -> Celery:
     google_channel_ttl = float(os.getenv('GOOGLE_CHANNEL_TTL_IN_SECONDS', SEVEN_DAYS_IN_SECONDS))
     google_channel_renew_interval = google_channel_ttl - ONE_DAY_IN_SECONDS
 
+    zoom_token_ttl = float(os.getenv('ZOOM_TOKEN_TTL_IN_SECONDS', SEVEN_DAYS_IN_SECONDS))
+    zoom_token_renew_interval = zoom_token_ttl - ONE_DAY_IN_SECONDS
+
     app = Celery('appointment')
 
     app.config_from_object({
@@ -68,6 +71,10 @@ def create_celery_app() -> Celery:
                 'task': 'appointment.tasks.google.renew_google_channels',
                 'schedule': google_channel_renew_interval,
             },
+            'refresh-zoom-tokens': {
+                'task': 'appointment.tasks.zoom.refresh_zoom_tokens',
+                'schedule': zoom_token_renew_interval,
+            },
         },
     })
 

backend/src/appointment/commands/refresh_zoom_tokens.py

@@ -0,0 +1,81 @@
+"""Renew expiring Zoom auth tokens.
+
+Run periodically (e.g., weekly) to ensure tokens don't expire.
+Zoom tokens typically last ~90 days, so weekly renewal keeps a buffer.
+"""
+
+import json
+import logging
+
+from ..database import repo, models
+from ..dependencies.database import get_engine_and_session
+from ..dependencies.zoom import get_zoom_client
+from ..main import _common_setup
+
+
+def run():
+    _common_setup()
+
+    _, SessionLocal = get_engine_and_session()
+    db = SessionLocal()
+    zoom_connections = []
+    refreshed = 0
+    failed = 0
+    cleaned = 0
+
+    try:
+        zoom_connections = repo.external_connection.get_zoom(db)
+
+        for connection in zoom_connections:
+            if not connection or not connection.token:
+                continue
+
+            owner_id = connection.owner_id
+            try:
+                subscriber = repo.subscriber.get(db, owner_id)
+                if not subscriber:
+                    # Orphaned external connection so let's just remove it.
+                    repo.external_connection.delete(db, connection)
+                    cleaned += 1
+                    continue
+
+                zoom_client = get_zoom_client(subscriber)
+                token = json.loads(connection.token)
+
+                # Force the OAuth session to attempt a refresh by marking the token as expired.
+                token['expires_in'] = -100
+                token['expires_at'] = 0
+                zoom_client.setup(subscriber.id, token)
+                zoom_client.get_me()
+
+                new_token = zoom_client.client.token
+                repo.external_connection.update_token_by_connection(
+                    db,
+                    json.dumps(new_token),
+                    connection,
+                )
+                repo.external_connection.update_status(
+                    db,
+                    connection,
+                    models.ExternalConnectionStatus.ok,
+                )
+                refreshed += 1
+            except Exception:
+                # Mark the external connection as error so the user can see
+                # it in the Settings UI and fix it manually when they log in
+                repo.external_connection.update_status(
+                    db,
+                    connection,
+                    models.ExternalConnectionStatus.error,
+                )
+
+                # TODO: Send an email to the Subscriber
+                # https://github.com/thunderbird/appointment/issues/1662
+                failed += 1
+    finally:
+        db.close()
+
+    logging.info(
+        f'[refresh_zoom_tokens] Zoom token refresh complete: '
+        f'{refreshed} refreshed, {failed} failed, {cleaned} cleaned, {len(zoom_connections)} total processed'
+    )

backend/src/appointment/controller/apis/zoom_client.py

@@ -39,26 +39,28 @@ def scopes(self):
         """Returns the appropriate scopes"""
         return self.SCOPES
 
-    def check_expiry(self, token: dict | None):
-        """Checks expires_at and if expired sets expires_in to a negative number to trigger refresh"""
+    def check_expiry(self, token: dict | None, threshold: float):
+        """Checks expires_at and sets expires_in to a negative number to trigger refresh
+           if already expired or within the given renewal threshold
+        """
         if not token:
             return token
 
         expires_at = token.get('expires_at')
-        if expires_at and expires_at <= time.time():
+        if expires_at and expires_at <= (time.time() + threshold):
             token['expires_in'] = -100
         elif not expires_at:
             # We shouldn't have to handle this but just in case alert us!
             sentry_sdk.capture_message('Expires at is missing!')
 
         return token
 
-    def setup(self, subscriber_id=None, token=None):
-        """Setup our oAuth session"""
+    def setup(self, subscriber_id=None, token=None, threshold=0.0):
+        """Setup our OAuth session"""
         if isinstance(token, str):
             token = json.loads(token)
 
-        token = self.check_expiry(token)
+        token = self.check_expiry(token, threshold)
 
         self.subscriber_id = subscriber_id
         self.client = OAuth2Session(

backend/src/appointment/database/repo/external_connection.py

@@ -5,6 +5,7 @@
 
 import logging
 import os
+from datetime import UTC, datetime
 import sentry_sdk
 from sqlalchemy.orm import Session
 from .. import models
@@ -36,7 +37,14 @@ def update_token(
     if db_results is None or len(db_results) == 0:
         return None
 
-    db_external_connection = db_results[0]
+    return update_token_by_connection(db, token, db_results[0])
+
+
+def update_token_by_connection(
+    db: Session,
+    token: str,
+    db_external_connection: models.ExternalConnections,
+):
     db_external_connection.token = token
     db.commit()
     db.refresh(db_external_connection)
@@ -54,6 +62,12 @@ def delete_by_type(db: Session, subscriber_id: int, type: models.ExternalConnect
     return True
 
 
+def delete(db: Session, db_external_connection: models.ExternalConnections):
+    db.delete(db_external_connection)
+    db.commit()
+    return True
+
+
 def get_by_type(
     db: Session, subscriber_id: int, type: models.ExternalConnectionType, type_id: str | None = None
 ) -> list[models.ExternalConnections] | None:
@@ -144,6 +158,18 @@ def update_name(db: Session, db_external_connection: models.ExternalConnections,
     return db_external_connection
 
 
+def update_status(
+    db: Session,
+    db_external_connection: models.ExternalConnections,
+    status: models.ExternalConnectionStatus,
+):
+    db_external_connection.status = status
+    db_external_connection.status_checked_at = datetime.now(UTC)
+    db.commit()
+    db.refresh(db_external_connection)
+    return db_external_connection
+
+
 def get_subscriber_without_oidc_by_email(db: Session, email: str):
     """Return a subscriber by the OIDC recovery email address without an OIDC connection"""
     # Subquery to check if a subscriber has an OIDC external connection
@@ -158,3 +184,15 @@ def get_subscriber_without_oidc_by_email(db: Session, email: str):
     query = db.query(models.Subscriber).filter(~oidc_exists).filter(models.Subscriber.email == email)
 
     return query.first()
+
+
+def get_zoom(db: Session) -> list[models.ExternalConnections] | None:
+    """Return all external connections by Zoom type"""
+    query = (
+        db.query(models.ExternalConnections)
+        .filter(models.ExternalConnections.type == models.ExternalConnectionType.zoom)
+    )
+
+    result = query.all()
+
+    return result

backend/src/appointment/routes/commands.py

@@ -69,3 +69,10 @@ def backfill_channels():
             backfill_google_channels.run()
     except FileExistsError:
         print('backfill-google-channels is already running, skipping.')
+
+@router.command('refresh-zoom-tokens')
+def refresh_tokens():
+    from ..tasks.zoom import refresh_zoom_tokens as refresh_zoom_tokens_task
+
+    refresh_zoom_tokens_task.delay()
+    print('refresh-zoom-tokens task queued.')

backend/src/appointment/tasks/__init__.py

@@ -1,2 +1,3 @@
 from appointment.tasks.health import *  # noqa: F401,F403
 from appointment.tasks.google import *  # noqa: F401,F403
+from appointment.tasks.zoom import *  # noqa: F401,F403

backend/src/appointment/tasks/locks.py

@@ -0,0 +1,72 @@
+import logging
+import os
+import uuid
+from contextlib import contextmanager
+from appointment.dependencies.database import get_redis
+
+DEFAULT_LOCK_TTL_SECONDS = 60 * 60
+
+
+class TaskLockFailed(Exception):
+    """Raised when a task lock cannot be acquired."""
+
+
+def _task_lock_key(task_name: str) -> str:
+    return f'lock:task:{task_name}'
+
+
+def acquire_task_lock(redis_instance, task_name: str, ttl_seconds: int = DEFAULT_LOCK_TTL_SECONDS) -> str | None:
+    lock_key = _task_lock_key(task_name)
+    lock_token = str(uuid.uuid4())
+    lock_acquired = bool(redis_instance.set(lock_key, lock_token, nx=True, ex=ttl_seconds))
+    if not lock_acquired:
+        return None
+    return lock_token
+
+
+def release_task_lock(redis_instance, task_name: str, lock_token: str):
+    """Only delete the lock if still owned by this task instance."""
+    lock_key = _task_lock_key(task_name)
+
+    with redis_instance.pipeline() as pipe:
+        pipe.watch(lock_key)
+        if pipe.get(lock_key) == lock_token:
+            pipe.multi()
+            pipe.delete(lock_key)
+            pipe.execute()
+        else:
+            pipe.unwatch()
+
+
+@contextmanager
+def task_lock(task_name: str, ttl_seconds: int = DEFAULT_LOCK_TTL_SECONDS):
+    """Context manager that acquires and releases a distributed task lock.
+
+    Raises TaskLockFailed if Redis is available but the lock is already held.
+    When Redis is unavailable, execution proceeds without a lock.
+    """
+    import sentry_sdk
+
+    redis_instance = get_redis(os.getenv('REDIS_CELERY_DB'))
+    lock_token = None
+
+    if redis_instance is not None:
+        lock_token = acquire_task_lock(redis_instance, task_name, ttl_seconds)
+        if lock_token is None:
+            raise TaskLockFailed(f'Failed to acquire lock for {task_name}')
+    else:
+        logging.warning(f'Redis unavailable; running {task_name} without distributed lock.')
+
+    try:
+        yield
+    except Exception as e:
+        logging.warning(f'{task_name} failed: {e}')
+        sentry_sdk.capture_exception(e)
+        raise
+    finally:
+        if lock_token is not None:
+            try:
+                release_task_lock(redis_instance, task_name, lock_token)
+            except Exception as e:
+                logging.warning(f'Failed to release {task_name} lock: {e}')
+                sentry_sdk.capture_exception(e)

backend/src/appointment/tasks/zoom.py

@@ -0,0 +1,22 @@
+import logging
+import sentry_sdk
+
+from appointment.celery_app import celery
+from appointment.tasks.locks import TaskLockFailed, task_lock
+
+
+@celery.task
+def refresh_zoom_tokens():
+    from appointment.commands.refresh_zoom_tokens import run
+
+    try:
+        with task_lock(refresh_zoom_tokens.__name__):
+            logging.info('Starting Zoom token check')
+            run()
+            logging.info('Zoom token check complete')
+    except TaskLockFailed:
+        logging.info('Zoom token check is already running, skipping.')
+    except Exception as e:
+        logging.error(f'Zoom token check failed: {e}')
+        sentry_sdk.capture_exception(e)
+        raise