Add 3CX roster sync Lambda with daily EventBridge trigger

Syncs DynamoDB roster from a configured 3CX group daily at 6am ET
(before the 7am schedule post). Includes ThreeCXClient for XAPI
authentication and group member queries. Preserves existing
slack_user_id links and guards against accidental roster wipes.

Closes #4

requirements.txt

@@ -1,2 +1,3 @@
 slack_bolt>=1.18.0,<2.0
 boto3>=1.28.0
+requests>=2.31.0

src/roster_sync.py

@@ -0,0 +1,123 @@
+"""Lambda handler — syncs the DynamoDB roster from 3CX group members.
+
+Runs daily via EventBridge. Pulls members from the configured 3CX group,
+filters to Extension type only (excludes RingGroups, IVRs, Voicemail, etc.),
+and syncs to DynamoDB. Preserves existing slack_user_id links.
+"""
+
+import logging
+import os
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+import boto3
+
+from src.three_cx_client import ThreeCXClient
+from src.schedule import ShiftSchedule
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+EASTERN = ZoneInfo("America/New_York")
+
+# System extensions to exclude from roster sync
+EXCLUDE_NAMES = {"Voicemail", "IVR", "Fax"}
+
+
+def get_3cx_credentials() -> dict:
+    ssm = boto3.client("ssm")
+    prefix = os.environ.get("TCX_SSM_PREFIX", "/3cx-scheduler")
+    params = ssm.get_parameters_by_path(Path=prefix, WithDecryption=True)
+    creds = {}
+    for p in params["Parameters"]:
+        key = p["Name"].split("/")[-1]
+        creds[key] = p["Value"]
+    return creds
+
+
+def handler(event, context):
+    now = datetime.now(EASTERN)
+
+    # DST guard — two EventBridge rules fire, only one is at 6am ET
+    if now.hour != 6:
+        logger.info("ET hour is %d, not 6am — skipping (wrong DST rule fired)", now.hour)
+        return {"skipped": True}
+
+    group_name = os.environ.get("SYNC_GROUP", "DEFAULT")
+
+    logger.info("Starting roster sync from 3CX group '%s' at %s", group_name, now.isoformat())
+
+    creds = get_3cx_credentials()
+    client = ThreeCXClient(
+        domain=creds["domain"],
+        auth_mode="oauth",
+        client_id=creds["client_id"],
+        client_secret=creds["client_secret"],
+    )
+
+    members = client.get_group_members(group_name)
+    # Only sync actual phone extensions, not ring groups, IVRs, etc.
+    extensions = [
+        m for m in members
+        if m.get("Type") == "Extension" and m.get("MemberName") not in EXCLUDE_NAMES
+    ]
+    logger.info("Found %d extensions in 3CX group '%s' (filtered from %d members)", len(extensions), group_name, len(members))
+
+    schedule = ShiftSchedule()
+    current_roster = {item["SK"]: item for item in schedule.get_roster()}
+
+    threecx_extensions = set()
+    added = []
+    updated = []
+
+    for ext in extensions:
+        number = str(ext.get("Number", ""))
+        name = ext.get("MemberName", "").strip()
+
+        if not number or not name:
+            continue
+
+        threecx_extensions.add(number)
+        existing = current_roster.get(number)
+
+        if existing:
+            if existing.get("name") != name:
+                schedule.table.update_item(
+                    Key={"PK": "ROSTER", "SK": number},
+                    UpdateExpression="SET #n = :name",
+                    ExpressionAttributeNames={"#n": "name"},
+                    ExpressionAttributeValues={":name": name},
+                )
+                updated.append(f"Ext {number}: {existing['name']} -> {name}")
+        else:
+            schedule.table.put_item(
+                Item={
+                    "PK": "ROSTER",
+                    "SK": number,
+                    "name": name,
+                    "extension": number,
+                    "slack_user_id": "",
+                }
+            )
+            added.append(f"Ext {number}: {name}")
+
+    removed = []
+    # Only remove departed employees if we got results from 3CX
+    # (prevents wiping roster on API errors or misconfigured group)
+    if threecx_extensions:
+        for ext_number, item in current_roster.items():
+            if ext_number not in threecx_extensions:
+                schedule.table.delete_item(Key={"PK": "ROSTER", "SK": ext_number})
+                removed.append(f"Ext {ext_number}: {item.get('name', 'Unknown')}")
+    elif current_roster:
+        logger.warning("No 3CX extensions found but roster is non-empty — skipping removal to prevent data loss")
+
+    result = {
+        "group": group_name,
+        "total_3cx": len(threecx_extensions),
+        "added": added,
+        "updated": updated,
+        "removed": removed,
+    }
+    logger.info("Roster sync complete: %s", result)
+    return result

src/three_cx_client.py

@@ -0,0 +1,133 @@
+import logging
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class ThreeCXClient:
+    """Client for 3CX V20 cloud-hosted management API (XAPI)."""
+
+    def __init__(self, domain: str, auth_mode: str = "user", **auth_kwargs):
+        """
+        Args:
+            domain: Your 3CX FQDN (e.g. "yourcompany.3cx.us")
+            auth_mode: "user" for extension login, "oauth" for Enterprise API client
+            auth_kwargs: credentials — see _authenticate_user / _authenticate_oauth
+        """
+        self.base_url = f"https://{domain}"
+        self.session = requests.Session()
+        self.session.headers.update({
+            "OData-Version": "4.0",
+            "Content-Type": "application/json",
+        })
+
+        if auth_mode == "oauth":
+            self._authenticate_oauth(auth_kwargs["client_id"], auth_kwargs["client_secret"])
+        else:
+            self._authenticate_user(auth_kwargs["username"], auth_kwargs["password"])
+
+    def _authenticate_user(self, username: str, password: str):
+        """Authenticate via extension/user credentials (any license tier)."""
+        resp = self.session.post(
+            f"{self.base_url}/webclient/api/Login/GetAccessToken",
+            json={"SecurityCode": "", "Username": username, "Password": password},
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        token = data.get("Token", {}).get("access_token") or data.get("access_token")
+        if not token:
+            raise ValueError(f"Failed to get access token. Response: {data}")
+        self.session.headers.update({"Authorization": f"Bearer {token}"})
+        logger.info("Authenticated to 3CX via user credentials")
+
+    def _authenticate_oauth(self, client_id: str, client_secret: str):
+        """Authenticate via OAuth2 client credentials (Enterprise license required).
+        API client must be created in 3CX Admin > Integrations > API."""
+        resp = self.session.post(
+            f"{self.base_url}/connect/token",
+            data={
+                "client_id": client_id,
+                "client_secret": client_secret,
+                "grant_type": "client_credentials",
+            },
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+        )
+        resp.raise_for_status()
+        token = resp.json()["access_token"]
+        self.session.headers.update({"Authorization": f"Bearer {token}"})
+        logger.info("Authenticated to 3CX via OAuth2 client credentials")
+
+    def get_group_members(self, group_name: str = "DEFAULT") -> list[dict]:
+        """Fetch members of a 3CX group (department).
+
+        Returns members with Number, MemberName, Type, etc.
+        """
+        # First, find the group ID by name
+        resp = self.session.get(
+            f"{self.base_url}/xapi/v1/Groups",
+            params={"$filter": f"Name eq '{group_name}'"},
+        )
+        resp.raise_for_status()
+        groups = resp.json().get("value", [])
+        if not groups:
+            logger.warning("Group '%s' not found in 3CX", group_name)
+            return []
+
+        group_id = groups[0]["Id"]
+
+        # Fetch members of that group
+        resp = self.session.get(f"{self.base_url}/xapi/v1/Groups({group_id})/Members")
+        resp.raise_for_status()
+        return resp.json().get("value", [])
+
+    def get_ring_group(self, extension_number: str) -> dict:
+        """Fetch ring group config by extension number."""
+        resp = self.session.get(
+            f"{self.base_url}/xapi/v1/RingGroups/Pbx.GetByNumber(number='{extension_number}')",
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    def update_ring_group_forwarding(
+        self,
+        ring_group_id: int,
+        closed_destination: str,
+        holiday_destination: str,
+    ):
+        """Update the OutOfOfficeRoute and HolidaysRoute on a ring group.
+
+        Args:
+            ring_group_id: Numeric ID from the ring group entity
+            closed_destination: Extension number for after-hours routing
+            holiday_destination: Extension number for holiday routing
+        """
+        payload = {
+            "OutOfOfficeRoute": {
+                "IsPromptEnabled": False,
+                "Route": {
+                    "To": "Extension",
+                    "Number": closed_destination,
+                    "External": "",
+                },
+            },
+            "HolidaysRoute": {
+                "IsPromptEnabled": False,
+                "Route": {
+                    "To": "Extension",
+                    "Number": holiday_destination,
+                    "External": "",
+                },
+            },
+        }
+        resp = self.session.patch(
+            f"{self.base_url}/xapi/v1/RingGroups({ring_group_id})",
+            json=payload,
+        )
+        resp.raise_for_status()
+        logger.info(
+            "Updated ring group %s: closed->Ext %s, holiday->Ext %s",
+            ring_group_id,
+            closed_destination,
+            holiday_destination,
+        )
+        return resp.status_code

template.yaml

@@ -122,6 +122,56 @@ Resources:
             Description: "Post weekly schedule Monday 7am EDT"
             Enabled: true
 
+  # --- Roster Sync Lambda (daily sync from 3CX) ---
+  RosterSyncFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      FunctionName: afterhours-roster-sync
+      Handler: src/roster_sync.handler
+      CodeUri: .
+      Timeout: 60
+      Environment:
+        Variables:
+          SHIFT_TABLE: !Ref ShiftTable
+          TCX_SSM_PREFIX: /3cx-scheduler
+          SYNC_GROUP: DEFAULT
+          TZ: !Ref Timezone
+      Policies:
+        - DynamoDBCrudPolicy:
+            TableName: !Ref ShiftTable
+        - Statement:
+            - Effect: Allow
+              Action:
+                - ssm:GetParametersByPath
+                - ssm:GetParameter
+                - ssm:GetParameters
+              Resource:
+                - !Sub "arn:aws:ssm:${AWS::Region}:${AWS::AccountId}:parameter/3cx-scheduler"
+                - !Sub "arn:aws:ssm:${AWS::Region}:${AWS::AccountId}:parameter/3cx-scheduler/*"
+            - Effect: Allow
+              Action:
+                - kms:Decrypt
+              Resource: "*"
+              Condition:
+                StringEquals:
+                  "kms:ViaService": !Sub "ssm.${AWS::Region}.amazonaws.com"
+      Events:
+        # Daily at 6am ET (before the 7am schedule post and 8am 3CX scheduler)
+        # EST: 6am ET = 11:00 UTC (Nov-Mar)
+        RosterSyncEST:
+          Type: Schedule
+          Properties:
+            Schedule: cron(0 11 ? * * *)
+            Description: "Sync roster from 3CX at 6am EST"
+            Enabled: true
+        # EDT: 6am ET = 10:00 UTC (Mar-Nov)
+        RosterSyncEDT:
+          Type: Schedule
+          Properties:
+            Schedule: cron(0 10 ? * * *)
+            Description: "Sync roster from 3CX at 6am EDT"
+            Enabled: true
+
 Outputs:
   SlackBotApiUrl:
     Description: URL for Slack app Request URL configuration
@@ -130,3 +180,5 @@ Outputs:
     Value: !Ref ShiftTable
   SlackBotFunctionArn:
     Value: !GetAtt SlackBotFunction.Arn
+  RosterSyncFunctionArn:
+    Value: !GetAtt RosterSyncFunction.Arn