front-integrations/src/user_sync/app.py

import json
import logging
import os
import time
from urllib import error, parse, request

import boto3
from google.oauth2 import service_account
from googleapiclient.discovery import build

logger = logging.getLogger()
logger.setLevel(logging.INFO)

FRONT_BASE = "https://api2.frontapp.com"
SCOPES = ["https://www.googleapis.com/auth/admin.directory.user.readonly"]
RATE_LIMIT_DELAY = 0.2

_sm = boto3.client("secretsmanager")
_front_token = None
_directory_service = None


def _get_front_token():
    global _front_token
    if _front_token is None:
        resp = _sm.get_secret_value(SecretId=os.environ["FRONT_SECRET_NAME"])
        _front_token = resp["SecretString"]
    return _front_token


def _get_directory_service():
    global _directory_service
    if _directory_service is None:
        resp = _sm.get_secret_value(SecretId=os.environ["GOOGLE_SECRET_NAME"])
        key_data = json.loads(resp["SecretString"])
        creds = service_account.Credentials.from_service_account_info(
            key_data,
            scopes=SCOPES,
            subject=os.environ["GOOGLE_ADMIN_EMAIL"],
        )
        _directory_service = build("admin", "directory_v1", credentials=creds)
    return _directory_service


def _list_google_users(org_units):
    service = _get_directory_service()
    all_users = []
    seen = set()

    for ou in org_units:
        logger.info("Fetching users from OU: %s", ou)
        page_token = None

        while True:
            result = (
                service.users()
                .list(
                    customer="my_customer",
                    maxResults=500,
                    projection="full",
                    orderBy="email",
                    query=f"orgUnitPath='{ou}'",
                    pageToken=page_token,
                )
                .execute()
            )

            for user in result.get("users", []):
                email = user.get("primaryEmail", "")
                if email not in seen:
                    seen.add(email)
                    all_users.append(user)

            page_token = result.get("nextPageToken")
            if not page_token:
                break

        logger.info(
            "  Found %d users in %s",
            len([u for u in all_users if u.get("primaryEmail") in seen]),
            ou,
        )

    return all_users


def _extract_user_fields(google_user):
    email = google_user.get("primaryEmail", "")

    job_title = ""
    orgs = google_user.get("organizations", [])
    if orgs:
        primary_org = next((o for o in orgs if o.get("primary")), orgs[0])
        job_title = primary_org.get("title", "")

    phone = ""
    phones = google_user.get("phones", [])
    if phones:
        work_phone = next(
            (p for p in phones if p.get("type") == "work" and p.get("value")), None
        )
        primary_phone = next(
            (p for p in phones if p.get("primary") and p.get("value")), None
        )
        fallback = next((p for p in phones if p.get("value")), None)
        phone = (work_phone or primary_phone or fallback or {}).get("value", "")

    return email, job_title, phone


def _front_request(method, url_path, body=None):
    url = f"{FRONT_BASE}{url_path}"
    headers = {
        "Authorization": f"Bearer {_get_front_token()}",
        "Accept": "application/json",
    }

    data = None
    if body is not None:
        data = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"

    req = request.Request(url, data=data, headers=headers, method=method)

    try:
        with request.urlopen(req) as resp:
            if resp.status == 204:
                return {"status": "updated"}
            return json.loads(resp.read().decode())
    except error.HTTPError as e:
        if e.code == 404:
            return None
        body_text = e.read().decode() if e.fp else ""
        raise RuntimeError(f"Front {method} {url_path} failed ({e.code}): {body_text}")


def _update_front_teammate(email, custom_fields):
    encoded = parse.quote(email, safe="")
    path = f"/teammates/alt:email:{encoded}"

    teammate = _front_request("GET", path)
    if teammate is None:
        return "not_in_front"

    merged = {**teammate.get("custom_fields", {}), **custom_fields}
    result = _front_request("PATCH", path, {"custom_fields": merged})

    if result and result.get("status") == "updated":
        return "updated"
    return "updated"


def handler(event, context):
    logger.info("Starting Front <- Google Directory sync")

    org_units = [ou.strip() for ou in os.environ["GOOGLE_OUS"].split(",") if ou.strip()]
    logger.info("Syncing OUs: %s", ", ".join(org_units))

    google_users = _list_google_users(org_units)
    logger.info("Found %d total users across %d OUs", len(google_users), len(org_units))

    summary = {
        "updated": 0,
        "not_in_front": 0,
        "no_data": 0,
        "errors": 0,
        "error_details": [],
    }

    for user in google_users:
        email, job_title, phone = _extract_user_fields(user)

        if not job_title and not phone:
            logger.info("Skipping %s - no title or phone in Google", email)
            summary["no_data"] += 1
            continue

        custom_fields = {}
        if job_title:
            custom_fields["Job Title"] = job_title
        if phone:
            custom_fields["Phone"] = phone

        try:
            result = _update_front_teammate(email, custom_fields)

            if result == "updated":
                logger.info("Updated %s: %s", email, json.dumps(custom_fields))
                summary["updated"] += 1
            elif result == "not_in_front":
                logger.info("Skipped %s - not a Front teammate", email)
                summary["not_in_front"] += 1

        except Exception as e:  # noqa: BLE001 - per-user isolation: one failed update must not abort the sync
            logger.error("Error updating %s: %s", email, e)
            summary["errors"] += 1
            summary["error_details"].append({"email": email, "message": str(e)})

        time.sleep(RATE_LIMIT_DELAY)

    logger.info("Sync summary: %s", json.dumps(summary))
    return {"statusCode": 200, "body": summary}