Use Google Calendar API calendar invites / Implement Push Notifications

backend/.env.example

@@ -10,6 +10,8 @@ APP_ENV=dev
 APP_ADMIN_ALLOW_LIST=
 APP_SETUP
 APP_ALLOW_FIRST_TIME_REGISTER=
+# Enable Google Calendar native invites (True/False). Defaults based on APP_ENV if unset.
+GOOGLE_INVITE_ENABLED=
 
 # -- BACKEND --
 BACKEND_URL=http://localhost:5000
@@ -95,6 +97,7 @@ GOOGLE_AUTH_CLIENT_ID=
 GOOGLE_AUTH_SECRET=
 GOOGLE_AUTH_PROJECT_ID=
 GOOGLE_AUTH_CALLBACK=http://localhost:5000/google/callback
+GOOGLE_CHANNEL_TTL_IN_SECONDS=604800 # 7 days
 
 # -- Zoom API --
 ZOOM_API_ENABLED=False

backend/.env.test

@@ -53,6 +53,7 @@ GOOGLE_AUTH_CLIENT_ID=
 GOOGLE_AUTH_SECRET=
 GOOGLE_AUTH_PROJECT_ID=
 GOOGLE_AUTH_CALLBACK=http://localhost:5000/google/callback
+GOOGLE_CHANNEL_TTL_IN_SECONDS=604800 # 7 days
 
 # -- TB ACCOUNTS --
 TB_ACCOUNTS_CALDAV_URL=https://stage-thundermail.com
@@ -70,6 +71,7 @@ SIGNED_SECRET=test-secret-pls-ignore
 SENTRY_DSN=
 # Possible values: prod, dev, test
 APP_ENV=test
+GOOGLE_INVITE_ENABLED=False
 # Possible values: password, fxa
 AUTH_SCHEME=password
 

backend/src/appointment/celery_app.py

@@ -3,6 +3,7 @@
 import sentry_sdk
 from celery import Celery
 from dotenv import load_dotenv
+from appointment.defines import ONE_DAY_IN_SECONDS, SEVEN_DAYS_IN_SECONDS
 
 
 def _base_redis_url() -> str:
@@ -40,6 +41,9 @@ def create_celery_app() -> Celery:
     sentry_sdk.set_extra('CELERY_RESULT_EXPIRES', result_expires)
     sentry_sdk.set_extra('CELERY_TASK_ALWAYS_EAGER', task_always_eager)
 
+    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
+
     app = Celery('appointment')
 
     app.config_from_object({
@@ -59,6 +63,10 @@ def create_celery_app() -> Celery:
                 'task': 'appointment.tasks.health.heartbeat',
                 'schedule': 60.0,
             },
+            'renew-google-channels': {
+                'task': 'appointment.tasks.google.renew_google_channels',
+                'schedule': google_channel_renew_interval,
+            },
         },
         'beat_schedule_filename': 'celerybeat-appointment-schedule',
     })

backend/src/appointment/commands/backfill_google_channels.py

@@ -0,0 +1,101 @@
+"""One-off command to set up Google Calendar watch channels for existing connected calendars."""
+
+import json
+import logging
+import uuid
+from datetime import datetime, timezone
+
+from google.oauth2.credentials import Credentials
+
+from ..controller.google_watch import get_webhook_url
+from ..database import repo, models
+from ..dependencies.database import get_engine_and_session
+from ..dependencies.google import get_google_client
+from ..main import _common_setup
+
+
+def run():
+    _common_setup()
+    google_client = get_google_client()
+
+    _, SessionLocal = get_engine_and_session()
+    db = SessionLocal()
+
+    webhook_url = get_webhook_url()
+    if not webhook_url:
+        print('BACKEND_URL not set, aborting.')
+        db.close()
+        return
+
+    # Find connected Google calendars that are the default in a schedule
+    # and don't yet have a watch channel
+    schedule_calendar_ids = [
+        s.calendar_id for s in
+        db.query(models.Schedule.calendar_id).filter(models.Schedule.calendar_id != None).all()  # noqa: E711
+    ]
+
+    all_calendars = db.query(models.Calendar).filter(
+        models.Calendar.provider == models.CalendarProvider.google,
+        models.Calendar.connected == True,  # noqa: E712
+        models.Calendar.id.in_(schedule_calendar_ids),
+    ).all()
+
+    candidates = []
+    for cal in all_calendars:
+        existing = repo.google_calendar_channel.get_by_calendar_id(db, cal.id)
+        if not existing:
+            candidates.append(cal)
+
+    print(f'Found {len(candidates)} connected Google calendar(s) without a watch channel.')
+
+    created = 0
+    skipped = 0
+    failed = 0
+
+    for calendar in candidates:
+        ext_conn = calendar.external_connection
+        if not ext_conn or not ext_conn.token:
+            print(f'  Calendar {calendar.id}: no external connection or token, skipping.')
+            skipped += 1
+            continue
+
+        try:
+            token = Credentials.from_authorized_user_info(
+                json.loads(ext_conn.token), google_client.SCOPES
+            )
+        except Exception as e:
+            print(f'  Calendar {calendar.id}: failed to parse token ({e}), skipping.')
+            skipped += 1
+            continue
+
+        try:
+            state = str(uuid.uuid4())
+            response = google_client.watch_events(calendar.user, webhook_url, token, state=state)
+            if not response:
+                print(f'  Calendar {calendar.id}: watch_events returned no response.')
+                failed += 1
+                continue
+
+            expiration_ms = int(response.get('expiration', 0))
+            expiration_dt = datetime.fromtimestamp(expiration_ms / 1000, tz=timezone.utc)
+
+            sync_token = google_client.get_initial_sync_token(calendar.user, token)
+
+            repo.google_calendar_channel.create(
+                db,
+                calendar_id=calendar.id,
+                channel_id=response['id'],
+                resource_id=response['resourceId'],
+                expiration=expiration_dt,
+                state=state,
+                sync_token=sync_token,
+            )
+            created += 1
+            print(f'  Calendar {calendar.id}: channel created (expires {expiration_dt}).')
+        except Exception as e:
+            print(f'  Calendar {calendar.id}: failed ({e}).')
+            logging.error(f'[backfill_google_channels] Error for calendar {calendar.id}: {e}')
+            failed += 1
+
+    db.close()
+    print(f'\nBackfill complete: {created} created, {skipped} skipped, {failed} failed.')

backend/src/appointment/commands/renew_google_channels.py

@@ -0,0 +1,87 @@
+"""Renew expiring Google Calendar push notification channels.
+
+Run periodically (e.g., daily) to ensure channels don't expire.
+Google channels typically last ~7 days, so daily renewal keeps a buffer.
+"""
+
+import json
+import logging
+import uuid
+from datetime import datetime, timedelta, timezone
+
+from google.oauth2.credentials import Credentials
+
+from ..controller.google_watch import get_webhook_url
+from ..database import repo
+from ..dependencies.database import get_engine_and_session
+from ..dependencies.google import get_google_client
+from ..main import _common_setup
+
+
+def run():
+    from ..tasks.google import stop_google_channel
+
+    _common_setup()
+    google_client = get_google_client()
+
+    _, SessionLocal = get_engine_and_session()
+    db = SessionLocal()
+
+    webhook_url = get_webhook_url()
+    if not webhook_url:
+        logging.error('[renew_google_channels] BACKEND_URL not set, aborting')
+        db.close()
+        return
+
+    # Renew channels that expire within the next 24 hours
+    threshold = datetime.now(tz=timezone.utc) + timedelta(hours=24)
+    channels = repo.google_calendar_channel.get_expiring(db, before=threshold)
+
+    renewed = 0
+    failed = 0
+
+    for channel in channels:
+        calendar = channel.calendar
+        if not calendar or not calendar.connected:
+            repo.google_calendar_channel.delete(db, channel)
+            continue
+
+        external_connection = calendar.external_connection
+        if not external_connection or not external_connection.token:
+            repo.google_calendar_channel.delete(db, channel)
+            continue
+
+        token = Credentials.from_authorized_user_info(json.loads(external_connection.token), google_client.SCOPES)
+
+        # Stop the old channel (fire-and-forget via Celery with retries)
+        stop_google_channel.delay(channel.channel_id, channel.resource_id, external_connection.token)
+
+        # Create a new channel
+        try:
+            new_state = str(uuid.uuid4())
+            response = google_client.watch_events(calendar.user, webhook_url, token, state=new_state)
+            if response:
+                expiration_ms = int(response.get('expiration', 0))
+                expiration_dt = datetime.fromtimestamp(expiration_ms / 1000, tz=timezone.utc)
+
+                repo.google_calendar_channel.update_expiration(
+                    db,
+                    channel,
+                    new_channel_id=response['id'],
+                    new_resource_id=response['resourceId'],
+                    new_expiration=expiration_dt,
+                    new_state=new_state,
+                )
+                renewed += 1
+            else:
+                repo.google_calendar_channel.delete(db, channel)
+                failed += 1
+        except Exception as e:
+            logging.error(f'[renew_google_channels] Failed to renew channel for calendar {calendar.id}: {e}')
+            failed += 1
+
+    db.close()
+    logging.info(
+        f'[renew_google_channels] Channel renewal complete: '
+        f'{renewed} renewed, {failed} failed, {len(channels)} total processed'
+    )